jsancio commented on code in PR #20318:
URL: https://github.com/apache/kafka/pull/20318#discussion_r2460760550
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -633,6 +634,51 @@ public void testBeginQuorumEpochHeartbeat(boolean
withKip853Rpc) throws Exceptio
context.assertSentBeginQuorumEpochRequest(epoch, Set.of(remoteId1,
remoteId2));
}
+ @ParameterizedTest
+ @ValueSource(booleans = { true, false })
+ public void testBeginQuorumShouldNotSendAfterFetchRequest(boolean
withKip853Rpc) throws Exception {
+ ReplicaKey localId = replicaKey(randomReplicaId(), true);
+ int remoteId1 = localId.id() + 1;
+ int remoteId2 = localId.id() + 2;
+ ReplicaKey replicaKey1 = replicaKey(remoteId1, withKip853Rpc);
+ ReplicaKey replicaKey2 = replicaKey(remoteId2, withKip853Rpc);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(
+ localId.id(), localId.directoryId().get())
+ .withKip853Rpc(withKip853Rpc)
+ .withStartingVoters(VoterSetTest.voterSet(Stream.of(localId,
replicaKey1, replicaKey2)), KRaftVersion.KRAFT_VERSION_1)
+ .build();
Review Comment:
Let's fix the formatting. Extra 4 spaces and multiple parameters in the same
line.
```java
RaftClientTestContext context = new RaftClientTestContext.Builder(
localId.id(),
localId.directoryId().get()
)
.withKip853Rpc(withKip853Rpc)
.withStartingVoters(
VoterSetTest.voterSet(Stream.of(localId, replicaKey1,
replicaKey2)),
KRaftVersion.KRAFT_VERSION_1
)
.build();
```
##########
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:
Okay. This algorithm works because the max fetch wait time is hardcoded to
500ms which is much smaller than `fetch timeout / 2` for most configurations.
https://github.com/apache/kafka/pull/20318/files#diff-1da15c51e641ea46ea5c86201ab8f21cfee9e7c575102a39c7bae0d5ffd7de39R260
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -633,6 +634,51 @@ public void testBeginQuorumEpochHeartbeat(boolean
withKip853Rpc) throws Exceptio
context.assertSentBeginQuorumEpochRequest(epoch, Set.of(remoteId1,
remoteId2));
}
+ @ParameterizedTest
+ @ValueSource(booleans = { true, false })
+ public void testBeginQuorumShouldNotSendAfterFetchRequest(boolean
withKip853Rpc) throws Exception {
+ ReplicaKey localId = replicaKey(randomReplicaId(), true);
Review Comment:
Make this name consistent with the rest of the names. This is the key and
not the id. E.g. `localKey`.
--
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]
