jsancio commented on a change in pull request #9476:
URL: https://github.com/apache/kafka/pull/9476#discussion_r510331249
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java
##########
@@ -1536,67 +1522,70 @@ public void
testObserverLeaderRediscoveryAfterRequestTimeout() throws Exception
int otherNodeId = 2;
int epoch = 5;
Set<Integer> voters = Utils.mkSet(leaderId, otherNodeId);
- KafkaRaftClient client = buildClient(voters);
- discoverLeaderAsObserver(client, voters, leaderId, epoch);
- pollUntilSend(client);
- RaftRequest.Outbound fetchRequest1 = assertSentFetchRequest();
+ RaftClientTestContext context = RaftClientTestContext.build(voters);
+
+ context.discoverLeaderAsObserver(voters, leaderId, epoch);
+
+ context.pollUntilSend();
+ RaftRequest.Outbound fetchRequest1 = context.assertSentFetchRequest();
assertEquals(leaderId, fetchRequest1.destinationId());
- assertFetchRequestData(fetchRequest1, epoch, 0L, 0);
+ RaftClientTestContext.assertFetchRequestData(fetchRequest1, epoch, 0L,
0);
- time.sleep(requestTimeoutMs);
- pollUntilSend(client);
+ context.time.sleep(REQUEST_TIMEOUT_MS);
+ context.pollUntilSend();
// We should retry the Fetch against the other voter since the original
// voter connection will be backing off.
- RaftRequest.Outbound fetchRequest2 = assertSentFetchRequest();
+ RaftRequest.Outbound fetchRequest2 = context.assertSentFetchRequest();
assertNotEquals(leaderId, fetchRequest2.destinationId());
assertTrue(voters.contains(fetchRequest2.destinationId()));
- assertFetchRequestData(fetchRequest2, epoch, 0L, 0);
+ RaftClientTestContext.assertFetchRequestData(fetchRequest2, epoch, 0L,
0);
- deliverResponse(fetchRequest2.correlationId,
fetchRequest2.destinationId(),
+ context.deliverResponse(fetchRequest2.correlationId,
fetchRequest2.destinationId(),
fetchResponse(epoch, leaderId, MemoryRecords.EMPTY, 0L,
Errors.FENCED_LEADER_EPOCH));
- client.poll();
+ context.client.poll();
- assertEquals(ElectionState.withElectedLeader(epoch, leaderId, voters),
quorumStateStore.readElectionState());
+ assertEquals(ElectionState.withElectedLeader(epoch, leaderId, voters),
context.quorumStateStore.readElectionState());
}
@Test
public void testLeaderGracefulShutdown() throws Exception {
int otherNodeId = 1;
- Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
int epoch = 1;
- KafkaRaftClient client = initializeAsLeader(voters, epoch);
+ Set<Integer> voters = Utils.mkSet(LOCAL_ID, otherNodeId);
+
+ RaftClientTestContext context =
RaftClientTestContext.initializeAsLeader(voters, epoch);
// Now shutdown
int shutdownTimeoutMs = 5000;
- CompletableFuture<Void> shutdownFuture =
client.shutdown(shutdownTimeoutMs);
+ CompletableFuture<Void> shutdownFuture =
context.client.shutdown(shutdownTimeoutMs);
// We should still be running until we have had a chance to send
EndQuorumEpoch
- assertTrue(client.isShuttingDown());
- assertTrue(client.isRunning());
+ assertTrue(context.client.isShuttingDown());
Review comment:
I thought about this last night and hack some solutions. I wasn't very
pleased with the result. Let's explore this improvement in a future PR.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
