kirktrue commented on code in PR #17700:
URL: https://github.com/apache/kafka/pull/17700#discussion_r1930979759
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java:
##########
@@ -3437,6 +3445,181 @@ public void testPollWithRedundantCreateFetchRequests() {
}
+ @Test
+ public void testFetchRequestWithBufferedPartitions() {
+ buildFetcher();
+
+ int numNodes = 2;
+ Set<TopicPartition> partitions = Set.of(tp0, tp1, tp2, tp3);
+ assignFromUser(partitions, numNodes);
+
+ // Seek each partition so that it becomes eligible to fetch.
+ partitions.forEach(tp -> subscriptions.seek(tp, 0));
+
+ List<Node> nodes = partitions.stream()
+ .map(tp -> metadata.fetch().leaderFor(tp))
+ .filter(Objects::nonNull)
+ .distinct()
+ .collect(Collectors.toList());
+ assertEquals(numNodes, nodes.size(), "The number of nodes in the
cluster is incorrect");
+
+ Node node0 = nodes.get(0);
+ Node node1 = nodes.get(1);
+ List<TopicPartition> node0Partitions = partitionsForNode(node0,
partitions);
+ List<TopicPartition> node1Partitions = partitionsForNode(node1,
partitions);
+
+ // sendFetches() call #1.
+ assertEquals(
+ 2,
+ sendFetches(),
+ "The first fetch should issue requests to node 0 or node 1 since
neither has buffered data"
+ );
+ prepareFetchResponses(node0, node0Partitions, buildRecords(0, 10, 1));
+ prepareFetchResponses(node1, node1Partitions, buildRecords(0, 10, 1));
+ networkClientDelegate.poll(time.timer(0));
+ assertEquals(4, fetcher.fetchBuffer.bufferedPartitions().size());
+ assertCollected(node0Partitions.remove(0), partitions);
+ assertEquals(3, fetcher.fetchBuffer.bufferedPartitions().size());
+
+ // sendFetches() call #2.
+ assertEquals(
+ 0,
+ sendFetches(),
+ "The second fetch shouldn't issue requests to either node 0 or
node 1 since they both have buffered data"
+ );
+ networkClientDelegate.poll(time.timer(0));
+ assertCollected(node1Partitions.remove(0), partitions);
+ assertEquals(2, fetcher.fetchBuffer.bufferedPartitions().size());
+
+ // sendFetches() call #3.
+ assertEquals(
+ 0,
+ sendFetches(),
+ "The third fetch shouldn't issue requests to either node 0 or node
1 since they both have buffered data"
+ );
+ networkClientDelegate.poll(time.timer(0));
+ assertCollected(node0Partitions.remove(0), partitions);
+ assertEquals(1, fetcher.fetchBuffer.bufferedPartitions().size());
+
+ // Node 0's partitions have all been collected, so validate that and
then reset the list of partitions
+ // from which to fetch data so the next pass should request can fetch
more data.
+ assertTrue(node0Partitions.isEmpty());
+ node0Partitions = partitionsForNode(node0, partitions);
+
+ // sendFetches() call #4.
+ assertEquals(
+ 1,
+ sendFetches(),
+ "The fourth fetch should issue a request to node 0 since its
buffered data was collected"
+ );
+ prepareFetchResponses(node0, node0Partitions, buildRecords(10, 10, 1));
+ networkClientDelegate.poll(time.timer(0));
+ assertCollected(node1Partitions.remove(0), partitions);
+ assertEquals(2, fetcher.fetchBuffer.bufferedPartitions().size());
+
+ // Node 1's partitions have likewise all been collected, so validate
that and reset.
+ assertTrue(node1Partitions.isEmpty());
+ node1Partitions = partitionsForNode(node1, partitions);
+
+ // sendFetches() call #5.
+ assertEquals(
+ 1,
+ sendFetches(),
+ "The fifth fetch should issue a request to node 1 since its
buffered data was collected"
+ );
+ prepareFetchResponses(node1, node1Partitions, buildRecords(10, 10, 1));
+ networkClientDelegate.poll(time.timer(0));
+ assertEquals(4, fetcher.fetchBuffer.bufferedPartitions().size());
+
+ assertEquals(
+ partitions,
+ fetchRecords().keySet(),
+ "Records from all partitions should have been collected"
+ );
+
+ assertEquals(
+ 0,
+ fetcher.fetchBuffer.bufferedPartitions().size(),
+ "There was still data remaining in the fetch buffer"
+ );
+
+ // sendFetches() call #6.
+ assertEquals(
+ 2,
+ sendFetches(),
+ "The sixth fetch should issue a request to nodes 0 and 1 since its
buffered data was collected"
+ );
+ prepareFetchResponses(node0, node0Partitions, buildRecords(20, 10, 1));
+ prepareFetchResponses(node1, node1Partitions, buildRecords(20, 10, 1));
+ networkClientDelegate.poll(time.timer(0));
+ assertEquals(4, fetcher.fetchBuffer.bufferedPartitions().size());
+
+ assertEquals(
+ partitions,
+ fetchRecords().keySet(),
+ "Records from all partitions should have been collected"
+ );
+
+ assertEquals(
+ 0,
+ fetcher.fetchBuffer.bufferedPartitions().size(),
+ "There was still data remaining in the fetch buffer"
+ );
+ }
+
+ private List<TopicPartition> partitionsForNode(Node node,
Set<TopicPartition> partitions) {
+ int nodesPerPartition = 2;
Review Comment:
Resolved.
--
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
