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

Code before:
    ) {
        long timeUntilNextBeginQuorumSend = state.timeUntilBeginQuorumEpochTimerExpires(currentTimeMs);
        if (timeUntilNextBeginQuorumSend == 0) {
            VoterSet voters = partitionState.lastVoterSet();
            Function<Integer, Node> nodeSupplier = voterId ->
                voters
                    .voterNode(voterId, channel.listenerName())
                    .orElseThrow(() ->
                        new IllegalStateException(
                            String.format(
                                "Unknown endpoint for voter id %d for listener name %s",
                                voterId,
                                channel.listenerName()
                            )
                        )
                    );

            Set<ReplicaKey> needToSendBeginQuorumRequests = state.needToSendBeginQuorumRequests(currentTimeMs);
            timeUntilNextBeginQuorumSend = maybeSendRequests(
                currentTimeMs,
                voters
                    .voterKeys()
                    .stream()
                    .filter(key -> key.id() != quorum.localIdOrThrow())
                    .filter(needToSendBeginQuorumRequests::contains)
                    .collect(Collectors.toUnmodifiableSet()),
                nodeSupplier,
                this::buildBeginQuorumEpochRequest
            );
            state.resetBeginQuorumEpochTimer(currentTimeMs);
        }
        return timeUntilNextBeginQuorumSend;
    }

    private long pollResigned(long currentTimeMs) {
        ResignedState state = quorum.resignedStateOrThrow();
        long endQuorumBackoffMs = maybeSendRequests(
            currentTimeMs,
            partitionState
                .lastVoterSet()
                .voterNodes(state.unackedVoters().stream(), channel.listenerName()),
            () -> buildEndQuorumEpochRequest(state)
        );

        GracefulShutdown shutdown = this.shutdown.get();
        final long stateTimeoutMs;
        if (shutdown != null) {
            // If the replica is shutting down, it will remain in the resigned state
            // until either the shutdown expires or an election bumps the epoch
            stateTimeoutMs = shutdown.remainingTimeMs();
        } else if (state.hasElectionTimeoutExpired(currentTimeMs)) {

Review comment (performance):
I think you can clean up and make this implementation more efficient by not calling `voters.voterKeys` and instead leverage the replica keys returned by `state.needToSendBeginQuorumRequests(currentTimeMs)`.

Code after:
            snapshotId,
            MAX_FETCH_SIZE_BYTES,
            snapshotSize
        );
    }

    private FetchSnapshotResponseData.PartitionSnapshot addQuorumLeader(
        FetchSnapshotResponseData.PartitionSnapshot partitionSnapshot
    ) {
        partitionSnapshot.currentLeader()
            .setLeaderEpoch(quorum.epoch())
            .setLeaderId(quorum.leaderIdOrSentinel());

        return partitionSnapshot;
    }

    public boolean isRunning() {
        GracefulShutdown gracefulShutdown = shutdown.get();
        return gracefulShutdown == null || !gracefulShutdown.isFinished();
    }

    public boolean isShuttingDown() {
        GracefulShutdown gracefulShutdown = shutdown.get();
        return gracefulShutdown != null && !gracefulShutdown.isFinished();
    }

    private void appendBatch(
        LeaderState<T> state,
        BatchAccumulator.CompletedBatch<T> batch,
        long appendTimeMs
    ) {
        try {
            int epoch = state.epoch();
            LogAppendInfo info = appendAsLeader(batch.data);
            OffsetAndEpoch offsetAndEpoch = new OffsetAndEpoch(info.lastOffset(), epoch);
            CompletableFuture<Long> future = appendPurgatory.await(
                offsetAndEpoch.offset() + 1, Integer.MAX_VALUE);

            future.whenComplete((commitTimeMs, exception) -> {
                if (exception != null) {
                    logger.debug(
                        "Failed to commit {} records up to last offset {}",
                        batch.numRecords,
                        offsetAndEpoch,
                        exception
                    );
                } else {
                    long elapsedTime = Math.max(0, commitTimeMs - appendTimeMs);
                    double elapsedTimePerRecord = (double) elapsedTime / batch.numRecords;
                    kafkaRaftMetrics.updateCommitLatency(elapsedTimePerRecord, appendTimeMs);
                    logger.debug("Completed commit of {} records up to last offset {}", batch.numRecords, offsetAndEpoch);