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

Code before:
     * 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();
        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()

Review comment (suggestion):
How about:
```java
return voterStates.values()
    .stream()
    .filter(
        state -> state.replicaKey.id() != localVoterNode.voterKey().id() &&
        currentTimeMs - state.lastFetchTimestamp >= beginQuorumEpochTimeoutM
    )
    .map(ReplicaState::replicaKey)
    .collect(Collectors.toUnmodifiableSet());
```

Code after:
     * 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())
            );
        }
        return remainingMs;
    }

    /**
     * Reset the checkQuorumTimer if we've received fetch/fetchSnapshot request from the majority of the voter
     *