jsancio commented on code in PR #14428: URL: https://github.com/apache/kafka/pull/14428#discussion_r1341771602
########## raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java: ########## @@ -485,6 +485,49 @@ public void testHandleBeginQuorumEpochAfterUserInitiatedResign() throws Exceptio context.listener.currentLeaderAndEpoch()); } + @Test + public void testLeaderShouldResignLeadershipIfNotGetFetchRequestFromMajorityVoters() throws Exception { Review Comment: Can we also tests the opposite. That the leader doesn't resign if the majority of the replicas (including the leader) have fetch in the last `fetchTimeoutMs`? Can we also add test(s) under `KafkaRaftClientSnapshotTest` that show that the leader also considers `FETCH_SNAPSHOT` requests for determining network connectivity? ########## raft/src/main/java/org/apache/kafka/raft/LeaderState.java: ########## @@ -55,6 +57,11 @@ public class LeaderState<T> implements EpochState { private final Map<Integer, ReplicaState> observerStates = new HashMap<>(); private final Logger log; private final BatchAccumulator<T> accumulator; + private final Set<Integer> fetchedVoters = new HashSet<>(); + private final Timer fetchTimer; + private final int fetchTimeoutMs; + // The majority number of the voters excluding the leader. Ex: 3 voters, the value will be 1 + private final int majority; Review Comment: Let's not cache this value. Let's always compute this from the size of `voterStates`. We can add a function for this. I am hoping to add support for dynamic voters soon and this is one less thing that I would have to change :smile: . ########## raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java: ########## @@ -965,6 +965,10 @@ private CompletableFuture<FetchResponseData> handleFetchRequest( } int replicaId = FetchRequest.replicaId(request); + + Optional<LeaderState<T>> state = quorum.maybeLeaderState(); + state.ifPresent(s -> s.maybeResetMajorityFollowerFetchTimeout(replicaId, currentTimeMs)); Review Comment: I am thinking that the leader should only update the timer on successful FETCH and FETCH_SNAPSHOT responses. If the majority of the replicas are not successfully replicating the log, the leader will not be able commit write operations (`RaftClient::scheduleAppend`) from the state machine. ########## raft/src/main/java/org/apache/kafka/raft/LeaderState.java: ########## @@ -66,7 +73,9 @@ protected LeaderState( Set<Integer> voters, Set<Integer> grantingVoters, BatchAccumulator<T> accumulator, - LogContext logContext + LogContext logContext, + Time time, + int fetchTimeoutMs Review Comment: Let's re-order this so that it better matches `CandidateState` and `FollowerState`. How about this order for the parameters: ```java protected LeaderState( Time time, int localId, int epoch, long epochStartOffset, Set<Integer> voters, Set<Integer> grantingVoters, BatchAccumulator<T> accumulator, int fetchTimeoutMs, LogContext logContext ) ``` ########## raft/src/main/java/org/apache/kafka/raft/LeaderState.java: ########## @@ -76,9 +85,37 @@ protected LeaderState( boolean hasAcknowledgedLeader = voterId == localId; this.voterStates.put(voterId, new ReplicaState(voterId, hasAcknowledgedLeader)); } + this.majority = voters.size() / 2; this.grantingVoters = Collections.unmodifiableSet(new HashSet<>(grantingVoters)); this.log = logContext.logger(LeaderState.class); this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null"); + this.fetchTimeoutMs = fetchTimeoutMs; + this.fetchTimer = time.timer(fetchTimeoutMs); + } + + public boolean hasMajorityFollowerFetchTimeoutExpired(long currentTimeMs) { + fetchTimer.update(currentTimeMs); + boolean isExpired = fetchTimer.isExpired(); + if (isExpired) { + log.info("Did not receive fetch request from the majority of the voters within {}ms. Current fetched voters are {}.", + fetchTimeoutMs, fetchedVoters); + } + return isExpired; + } + + public void maybeResetMajorityFollowerFetchTimeout(int id, long currentTimeMs) { + updateFetchedVoters(id); + if (fetchedVoters.size() >= majority) { + fetchedVoters.clear(); + fetchTimer.update(currentTimeMs); + fetchTimer.reset(fetchTimeoutMs); + } + } + + private void updateFetchedVoters(int id) { + if (isVoter(id)) { + fetchedVoters.add(id); + } Review Comment: I see. I think that the invariant that we need to hold is that for any time span of `fetchTimeoutMs` the majority of the replicas have performed a successful `FETCH` and `FETCH_SNAPSHOT`. Note that `ReplicaState` already contains the `lastFetchTimestamp`. The part that is not clear to me is when or how to wake up the leader for a `poll`. We need to update `KafkaRaftClient::pollLeader` so that the replicas' last fetch time is taken into account when blocking on the `messageQueue.poll`. What do you think? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org