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

Code before:
                 * Local replica is not in the voter set because the replica is an observer.
                 */
                final var oldVoter = voters.voterKeys()
                    .stream()
                    .filter(replicaKey -> replicaKey.id() == localReplicaKey.id())
                    .findFirst()
                    .get();
                sendResult = maybeSendRemoveVoterRequest(state, oldVoter, currentTimeMs);
            } else {
                sendResult = maybeSendAddVoterRequest(state, currentTimeMs);
            }
            backoffMs = sendResult.timeToWaitMs();
            if (sendResult.requestSent()) {
                state.resetUpdateVoterSetPeriod(currentTimeMs);
            }
        } else {
            backoffMs = maybeSendFetchToBestNode(state, currentTimeMs);
        }
        if (shouldSendAddOrRemoveVoterRequest()) {
            return Math.min(
                backoffMs,
                state.remainingUpdateVoterSetPeriodMs(currentTimeMs)
            );
        } else {
            return backoffMs;
        }
    }

    private long maybeSendFetchToBestNode(FollowerState state, long currentTimeMs) {
        // If the current leader is backing off due to some failure or if the
        // request has timed out, then we attempt to send the Fetch to another
        // voter in order to discover if there has been a leader change.
        final long backoffMs;
        Node leaderNode = state.leaderNode(channel.listenerName());
        if (requestManager.hasRequestTimedOut(leaderNode, currentTimeMs)) {
            // Once the request has timed out backoff the connection
            requestManager.reset(leaderNode);
            backoffMs = maybeSendFetchToAnyBootstrap(currentTimeMs);
        } else if (requestManager.isBackingOff(leaderNode, currentTimeMs)) {
            backoffMs = maybeSendFetchToAnyBootstrap(currentTimeMs);
        } else if (!requestManager.hasAnyInflightRequest(currentTimeMs)) {
            backoffMs = maybeSendFetchOrFetchSnapshot(state, currentTimeMs);
        } else {
            backoffMs = requestManager.backoffBeforeAvailableBootstrapServer(currentTimeMs);
        }
        return Math.min(backoffMs, state.remainingFetchTimeMs(currentTimeMs));
    }

    private long maybeSendFetchOrFetchSnapshot(FollowerState state, long currentTimeMs) {
        final Supplier<ApiMessage> requestSupplier;


Review comment (suggestion):
Can you change the implementation so that `shouldSendAddOrRemoveVoterRequest` is only evaluated once?

You can also change the back off computation to something like:
```java
    return Math.min(
        backoffMs,
        shouldSendAddOrRemoveVoterRequest ?
            state.remainingUpdateVoterSetPeriodMs(currentTimeMs) :
            Integer.MAX_VALUE
    );
```

Code after:
                sendResult = maybeSendRemoveVoterRequest(state, oldVoter, currentTimeMs);
            } else {
                sendResult = maybeSendAddVoterRequest(state, currentTimeMs);
            }
            if (sendResult.requestSent()) {
                state.resetUpdateVoterSetPeriod(currentTimeMs);
            }
            return sendResult.timeToWaitMs();
        } else {
            return maybeSendFetchToBestNode(state, currentTimeMs);
        }
    }

    private long maybeSendFetchToBestNode(FollowerState state, long currentTimeMs) {
        // If the current leader is backing off due to some failure or if the
        // request has timed out, then we attempt to send the Fetch to another
        // voter in order to discover if there has been a leader change.
        final long backoffMs;
        Node leaderNode = state.leaderNode(channel.listenerName());
        if (requestManager.hasRequestTimedOut(leaderNode, currentTimeMs)) {
            // Once the request has timed out backoff the connection
            requestManager.reset(leaderNode);
            backoffMs = maybeSendFetchToAnyBootstrap(currentTimeMs);
        } else if (requestManager.isBackingOff(leaderNode, currentTimeMs)) {
            backoffMs = maybeSendFetchToAnyBootstrap(currentTimeMs);
        } else if (!requestManager.hasAnyInflightRequest(currentTimeMs)) {
            backoffMs = maybeSendFetchOrFetchSnapshot(state, currentTimeMs);
        } else {
            backoffMs = requestManager.backoffBeforeAvailableBootstrapServer(currentTimeMs);
        }
        return Math.min(backoffMs, state.remainingFetchTimeMs(currentTimeMs));
    }

    private long maybeSendFetchOrFetchSnapshot(FollowerState state, long currentTimeMs) {
        final Supplier<ApiMessage> requestSupplier;

        if (state.fetchingSnapshot().isPresent()) {
            RawSnapshotWriter snapshot = state.fetchingSnapshot().get();
            long snapshotSize = snapshot.sizeInBytes();

            requestSupplier = () -> buildFetchSnapshotRequest(snapshot.snapshotId(), snapshotSize);
        } else {
            requestSupplier = this::buildFetchRequest;
        }

        return maybeSendRequest(
            currentTimeMs,
            state.leaderNode(channel.listenerName()),
            requestSupplier
        ).timeToWaitMs();
    }