private long maybeSendBeginQuorumEpochRequests(
        LeaderState<T> state,
        long currentTimeMs
    ) {
        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()
                            )
                        )
                    );

            timeUntilNextBeginQuorumSend = maybeSendRequest(
                currentTimeMs,
                voters
                    .voterKeys()
                    .stream()
                    .filter(key -> key.id() != quorum.localIdOrThrow())
                    .collect(Collectors.toSet()),
                nodeSupplier,
                this::buildBeginQuorumEpochRequest
            );
            state.resetBeginQuorumEpochTimer(currentTimeMs);
        }
        return timeUntilNextBeginQuorumSend;
    }
Attempt to send a request. Return the time to wait before the request can be retried.
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java