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

Code before:
    }

    /**
     * Identifies voter replicas that require a {@code BeginQuorumEpoch} request from the leader.
     * <p>
     * The leader periodically checks whether any voter replicas have fallen behind in fetching
     * from it. This method evaluates all known voter replica states and determines which replicas
     * should receive a {@code BeginQuorumEpoch} request.
     * <p>
     * A replica will be included in the returned set if:
     * <ul>
     *   <li>It is not the local voter (the leader itself), and</li>
     *   <li>The time since its last fetch is greater than or equal to
     *       {@code beginQuorumEpochTimeoutMs}.</li>
     * </ul>
     * <p>
     * The {@link #beginQuorumEpochTimer} is updated with the current time before performing the check.
     * The returned set is unmodifiable to prevent external modification.
     *
     * @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) {
        Set<ReplicaKey> replicaKeys = new HashSet<>();
        beginQuorumEpochTimer.update(currentTimeMs);
        for (ReplicaState state : voterStates.values()) {
            if (state.replicaKey.id() != localVoterNode.voterKey().id()
                && currentTimeMs - state.lastFetchTimestamp >= beginQuorumEpochTimeoutMs) {
                replicaKeys.add(state.replicaKey());
            }
        }
        return Collections.unmodifiableSet(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();

Review comment (suggestion):
Do you need to update the timer? For example, the timer is never evaluated by this method. In this implementation the timer is updated and evaluated before calling this method [here](https://github.com/apache/kafka/pull/20318/files#diff-1da15c51e641ea46ea5c86201ab8f21cfee9e7c575102a39c7bae0d5ffd7de39R3107): `long timeUntilNextBeginQuorumSend = state.timeUntilBeginQuorumEpochTimerExpires(currentTimeMs);`

Code after:

    /**
     * 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;
        }
        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. " +
                "Current fetched voters are {}, and voters are {}",
                checkQuorumTimeoutMs,
                fetchedVoters,
                voterStates.values()
                    .stream()
                    .map(voter -> voter.replicaKey)
                    .collect(Collectors.toUnmodifiableSet())
            );