ahuang98 commented on code in PR #17352:
URL: https://github.com/apache/kafka/pull/17352#discussion_r1876623805
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java:
##########
@@ -2628,6 +2697,100 @@ public void
testVoteResponseIgnoredAfterBecomingFollower(boolean withKip853Rpc)
context.assertElectedLeader(epoch, voter3);
}
+ @ParameterizedTest
+ @ValueSource(booleans = { true, false })
+ public void
testFollowerLeaderRediscoveryAfterBrokerNotAvailableError(boolean
withKip853Rpc) throws Exception {
+ int localId = randomReplicaId();
+ int leaderId = localId + 1;
+ int otherNodeId = localId + 2;
+ int epoch = 5;
+ Set<Integer> voters = Utils.mkSet(leaderId, localId, otherNodeId);
+ List<InetSocketAddress> bootstrapServers = voters
+ .stream()
+ .map(RaftClientTestContext::mockAddress)
+ .collect(Collectors.toList());
+
+ RaftClientTestContext context = new
RaftClientTestContext.Builder(localId, voters)
+ .withBootstrapServers(Optional.of(bootstrapServers))
+ .withKip853Rpc(withKip853Rpc)
+ .withElectedLeader(epoch, leaderId)
+ .build();
+
+ context.pollUntilRequest();
+ RaftRequest.Outbound fetchRequest1 = context.assertSentFetchRequest();
+ assertEquals(leaderId, fetchRequest1.destination().id());
+ context.assertFetchRequestData(fetchRequest1, epoch, 0L, 0);
+
+ context.deliverResponse(
+ fetchRequest1.correlationId(),
+ fetchRequest1.destination(),
+ context.fetchResponse(epoch, -1, MemoryRecords.EMPTY, -1,
Errors.BROKER_NOT_AVAILABLE)
+ );
+ context.pollUntilRequest();
+
+ // We should retry the Fetch against the other voter since the original
+ // voter connection will be backing off.
+ RaftRequest.Outbound fetchRequest2 = context.assertSentFetchRequest();
+ assertNotEquals(leaderId, fetchRequest2.destination().id());
+
assertTrue(context.bootstrapIds.contains(fetchRequest2.destination().id()));
+ context.assertFetchRequestData(fetchRequest2, epoch, 0L, 0);
+
+ Errors error = fetchRequest2.destination().id() == leaderId ?
+ Errors.NONE : Errors.NOT_LEADER_OR_FOLLOWER;
+ context.deliverResponse(
+ fetchRequest2.correlationId(),
+ fetchRequest2.destination(),
+ context.fetchResponse(epoch, leaderId, MemoryRecords.EMPTY, 0L,
error)
+ );
+ context.client.poll();
+
+ context.assertElectedLeader(epoch, leaderId);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = { true, false })
+ public void testFollowerLeaderRediscoveryAfterRequestTimeout(boolean
withKip853Rpc) throws Exception {
+ int localId = randomReplicaId();
+ int leaderId = localId + 1;
+ int otherNodeId = localId + 2;
+ int epoch = 5;
+ Set<Integer> voters = Utils.mkSet(leaderId, localId, otherNodeId);
+ List<InetSocketAddress> bootstrapServers = voters
+ .stream()
+ .map(RaftClientTestContext::mockAddress)
+ .collect(Collectors.toList());
+
+ RaftClientTestContext context = new
RaftClientTestContext.Builder(localId, voters)
+ .withBootstrapServers(Optional.of(bootstrapServers))
+ .withKip853Rpc(withKip853Rpc)
+ .withElectedLeader(epoch, leaderId)
+ .build();
+
+ context.pollUntilRequest();
+ RaftRequest.Outbound fetchRequest1 = context.assertSentFetchRequest();
+ assertEquals(leaderId, fetchRequest1.destination().id());
+ context.assertFetchRequestData(fetchRequest1, epoch, 0L, 0);
+
+ context.time.sleep(context.requestTimeoutMs());
+ context.pollUntilRequest();
+
+ // We should retry the Fetch against the other voter since the original
+ // voter connection will be backing off.
+ RaftRequest.Outbound fetchRequest2 = context.assertSentFetchRequest();
+ assertNotEquals(leaderId, fetchRequest2.destination().id());
+
assertTrue(context.bootstrapIds.contains(fetchRequest2.destination().id()));
+ context.assertFetchRequestData(fetchRequest2, epoch, 0L, 0);
+
+ context.deliverResponse(
+ fetchRequest2.correlationId(),
+ fetchRequest2.destination(),
+ context.fetchResponse(epoch, leaderId, MemoryRecords.EMPTY, 0L,
Errors.FENCED_LEADER_EPOCH)
Review Comment:
hm, perhaps `testObserverLeaderRediscoveryAfterRequestTimeout` needs to
change too
--
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
