ahuang98 commented on code in PR #17352:
URL: https://github.com/apache/kafka/pull/17352#discussion_r1876616255
##########
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);
Review Comment:
yeah, I see how this doesn't add much value. I honestly could remove
everything from L2737 down (no need to even deliver response, we just want to
confirm if leader is unresponsive, we fetch from another node)
--
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
