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