Repository: apache/kafka
File: raft/src/main/java/org/apache/kafka/raft/LeaderState.java (Java)

Code before:
                    )
                );
            kraftVersionUpgradeState.set(new KRaftVersionUpgrade.Voters(updatedVoters));
        }
    }

    public long timeUntilBeginQuorumEpochTimerExpires(long currentTimeMs) {
        beginQuorumEpochTimer.update(currentTimeMs);
        return beginQuorumEpochTimer.remainingMs();
    }

    public void resetBeginQuorumEpochTimer(long currentTimeMs) {
        beginQuorumEpochTimer.update(currentTimeMs);
        beginQuorumEpochTimer.reset(beginQuorumEpochTimeoutMs);
    }

    public void updateLastReceivedFetchRequest(ReplicaKey replicaKey, long currentTimeMs) {
        beginQuorumEpochTimer.update(currentTimeMs);
        lastFetchRequestMs.put(replicaKey, beginQuorumEpochTimer.currentTimeMs());
    }

    public Set<ReplicaKey> needSendBeginQuorumRequestNodes(long currentTimeMs) {
        Set<ReplicaKey> replicaKeys = new HashSet<>();
        beginQuorumEpochTimer.update(currentTimeMs);
        for (Map.Entry<ReplicaKey, Long> entry : lastFetchRequestMs.entrySet()) {
            if (beginQuorumEpochTimer.currentTimeMs() - entry.getValue() >= beginQuorumEpochTimeoutMs) {
                replicaKeys.add(entry.getKey());
            }
        }
        return replicaKeys;
    }

    /**
     * Get the remaining time in milliseconds until the checkQuorumTimer expires.
     *
     * This will happen if we didn't receive a valid fetch/fetchSnapshot request from the majority
     * of the voters within checkQuorumTimeoutMs.
     *
     * @param currentTimeMs the current timestamp in millisecond
     * @return the remainingMs before the checkQuorumTimer expired
     */
    public long timeUntilCheckQuorumExpires(long currentTimeMs) {
        // if there's only 1 voter, it should never get expired.
        if (voterStates.size() == 1) {
            return Long.MAX_VALUE;
        }
        checkQuorumTimer.update(currentTimeMs);
        long remainingMs = checkQuorumTimer.remainingMs();
        if (remainingMs == 0) {
            log.info(
                "Did not receive fetch request from the majority of the voters within {}ms. " +

Review comment (suggestion):
From the code above at KafkaRaftClient#L3059,

```java
.filter(key -> !needToSendBeginQuorumRequest.contains(key))
```

 I think the condition should be changed to:
```suggestion
            if (beginQuorumEpochTimer.currentTimeMs() - entry.getValue() < beginQuorumEpochTimeoutMs) {
```

and we should rename the method to `skipSendBeginQuorumRequestNodes`.

Code after:
                );
            kraftVersionUpgradeState.set(new KRaftVersionUpgrade.Voters(updatedVoters));
        }
    }

    public long timeUntilBeginQuorumEpochTimerExpires(long currentTimeMs) {
        beginQuorumEpochTimer.update(currentTimeMs);
        return beginQuorumEpochTimer.remainingMs();
    }

    public void resetBeginQuorumEpochTimer(long currentTimeMs) {
        beginQuorumEpochTimer.update(currentTimeMs);
        beginQuorumEpochTimer.reset(beginQuorumEpochTimeoutMs);
    }

    /**
     * Determines the set of replicas that should receive a {@code BeginQuorumEpoch} request
     * based on the elapsed time since their last fetch.
     * <p>
     * For each remote voter (excluding the local node), if the time since the last
     * fetch exceeds the configured {@code beginQuorumEpochTimeoutMs}, the replica
     * is considered to need a new quorum epoch request.
     *
     * @param currentTimeMs the current system time in milliseconds
     * @return an unmodifiable set of {@link ReplicaKey} objects representing replicas
     *         that need to receive a {@code BeginQuorumEpoch} request
     */
    public Set<ReplicaKey> needToSendBeginQuorumRequests(long currentTimeMs) {
        return voterStates.values()
            .stream()
            .filter(
                state -> state.replicaKey.id() != localVoterNode.voterKey().id() &&
                currentTimeMs - state.lastFetchTimestamp >= beginQuorumEpochTimeoutMs
            )
            .map(ReplicaState::replicaKey)
            .collect(Collectors.toUnmodifiableSet());
    }

    /**
     * Get the remaining time in milliseconds until the checkQuorumTimer expires.
     *
     * This will happen if we didn't receive a valid fetch/fetchSnapshot request from the majority
     * of the voters within checkQuorumTimeoutMs.
     *
     * @param currentTimeMs the current timestamp in millisecond
     * @return the remainingMs before the checkQuorumTimer expired
     */
    public long timeUntilCheckQuorumExpires(long currentTimeMs) {
        // if there's only 1 voter, it should never get expired.
        if (voterStates.size() == 1) {
            return Long.MAX_VALUE;