TaiJuWu commented on code in PR #20318:
URL: https://github.com/apache/kafka/pull/20318#discussion_r2466050010
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -188,6 +188,29 @@ public void resetBeginQuorumEpochTimer(long currentTimeMs)
{
beginQuorumEpochTimer.reset(beginQuorumEpochTimeoutMs);
}
+ /**
+ * Determines the set of replicas that should receive a {@code
BeginQuorumEpoch} request
+ * based on the elapsed time since their last fetch.
+ * <p>
+ * For each remote voter (excluding the local node), if the time since the
last
+ * fetch exceeds the configured {@code beginQuorumEpochTimeoutMs}, the
replica
+ * is considered to need a new quorum epoch request.
+ *
+ * @param currentTimeMs the current system time in milliseconds
+ * @return an unmodifiable set of {@link ReplicaKey} objects representing
replicas
+ * that need to receive a {@code BeginQuorumEpoch} request
+ */
+ public Set<ReplicaKey> needToSendBeginQuorumRequests(long currentTimeMs) {
+ return voterStates.values()
+ .stream()
+ .filter(
+ state -> state.replicaKey.id() !=
localVoterNode.voterKey().id() &&
+ currentTimeMs - state.lastFetchTimestamp >=
beginQuorumEpochTimeoutMs
Review Comment:
> We already have that requirement today. Meaning the the fetch timeout
should be greater than 2 times the hard-coded max wait time.
If nobody can handle this, may I have chance to handle this?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
