private void pollListeners() {
        // Apply all of the pending registration
        while (true) {
            Registration<T> registration = pendingRegistrations.poll();
            if (registration == null) {
                break;
            }

            processRegistration(registration);
        }

        // Check listener progress to see if reads are expected
        quorum.highWatermark().ifPresent(highWatermarkMetadata ->
            updateListenersProgress(highWatermarkMetadata.offset())
        );

        // Notify the new listeners of the latest leader and epoch
        Optional<LeaderState<T>> leaderState = quorum.maybeLeaderState();
        if (leaderState.isPresent()) {
            maybeFireLeaderChange(leaderState.get());
        } else {
            maybeFireLeaderChange();
        }
    }
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