(kafka) branch trunk updated: MINOR: extend fetch session test to avoid false positive (#22734)

[li...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lianetm pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 624ca392efd MINOR: extend fetch session test to avoid false positive (#22734)
624ca392efd is described below

commit 624ca392efd0a67dd6386d7776f3eba1044e0f1e
Author: Lianet Magrans <98...@users.noreply.github.com>
AuthorDate: Thu Jul 2 13:07:31 2026 -0400

    MINOR: extend fetch session test to avoid false positive (#22734)
    
    Extend test to ensure there is an inflight request before completing.
    Final validation is about getting no records, so extend to ensure the
    case is a request sent an no records returned (and not that there was no
    request sent at all, which would lead to no records returned too but
    would be a false positive)
    
    Reviewers: TengYao Chi <fr...@apache.org>
---
 .../clients/consumer/internals/FetchRequestManagerTest.java    | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java
index 4a5dafaaeb8..4767b607417 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java
@@ -1771,7 +1771,12 @@ public class FetchRequestManagerTest {
         subscriptions.seek(tp0, 0);
 
         assertEquals(1, sendFetches());
+        networkClientDelegate.poll(time.timer(0));
+        assertEquals(1, client.inFlightRequestCount());
+        assertFalse(fetcher.hasCompletedFetches());
 
+        // Respond to the in-flight request with a response that includes an unexpected
+        // partition (tp1) that is not part of the fetch session.
         Map<TopicIdPartition, FetchResponseData.PartitionData> partitions = new LinkedHashMap<>();
         partitions.put(tidp0, new FetchResponseData.PartitionData()
                 .setPartitionIndex(tp0.partition())
@@ -1783,9 +1788,12 @@ public class FetchRequestManagerTest {
                 .setHighWatermark(100L)
                 .setLogStartOffset(0)
                 .setRecords(records));
-        client.prepareResponse(FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, new LinkedHashMap<>(partitions), List.of()));
+        client.respond(FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, new LinkedHashMap<>(partitions), List.of()));
         networkClientDelegate.poll(time.timer(0));
 
+        // The in-flight request has completed, but FetchSessionHandler rejected the whole
+        // response because of the unexpected partition (tp1), so nothing is buffered or
+        // returned to the consumer.
         assertEquals(0, client.inFlightRequestCount());
         assertFalse(fetcher.hasCompletedFetches());
         assertTrue(fetchRecords().isEmpty());