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 7c010c75839 KAFKA-20733: Fix
fetchResponseWithUnexpectedPartitionIsIgnored passing for wrong reason with
CONSUMER protocol (#22651)
7c010c75839 is described below
commit 7c010c75839dd71ff1d7b1f1f2c7434cbee47e91
Author: Park Jiwon <[email protected]>
AuthorDate: Tue Jun 30 01:16:14 2026 +0900
KAFKA-20733: Fix fetchResponseWithUnexpectedPartitionIsIgnored passing for
wrong reason with CONSUMER protocol (#22651)
## Summary
`KafkaConsumerTest.fetchResponseWithUnexpectedPartitionIsIgnored` was
annotated with `@EnumSource(GroupProtocol.class)`, but the CONSUMER
protocol case passed for the wrong reason.
The CONSUMER protocol uses `FetchRequestManager` and relies on
`ConsumerGroupHeartbeat` for rebalancing. Since the test does not set up
heartbeat-based rebalancing, partition assignment never completes and 0
records are returned — not because `FetchSessionHandler` correctly
rejected the response.
## Analysis
While writing the CONSUMER protocol test, we first attempted to use
`AsyncKafkaConsumerTest` as a reference. However, it was not suitable
because `fetchCollector` is replaced with a Mock, meaning
`FetchSessionHandler`'s actual rejection logic is never exercised.
We also found that naively adding the CONSUMER case to
`KafkaConsumerTest` does not work either. The test uses
`prepareResponse()` for the Join and Sync group requests, but these
responses were never consumed — because the CONSUMER protocol does not
go through the classic Join/Sync rebalance flow. As a result, the fetch
phase was never reached. `FetchRequestManagerTest` is the correct
location for this test, as it directly runs `FetchSessionHandler`
without mocking.
## Changes
- **`KafkaConsumerTest`**: Restrict
`fetchResponseWithUnexpectedPartitionIsIgnored` to `CLASSIC` protocol
only, where the existing setup correctly exercises
`FetchSessionHandler`'s unexpected partition rejection logic.
- **`FetchRequestManagerTest`**: Add
`testFetchResponseWithUnexpectedPartitionIsIgnored` to properly cover
the CONSUMER protocol path. The test verifies that when a
`FetchResponse` contains a partition not included in the `FetchRequest`,
`FetchSessionHandler` rejects the entire response and no records are
returned to the consumer.
- +**`KafkaConsumerTest`**: Rename
`fetchResponseWithUnexpectedPartitionIsIgnored`to
`testFetchResponseWithUnexpectedPartitionIsIgnored` to align with the
existing test naming convention in the class.
Reviewers: Lianet Magrans <[email protected]>
---
I'm happy to make any adjustments based on your feedback. Thank you to
the maintainers for taking the time to review this contribution!
---
.../kafka/clients/consumer/KafkaConsumerTest.java | 6 +++--
.../internals/FetchRequestManagerTest.java | 31 ++++++++++++++++++++++
2 files changed, 35 insertions(+), 2 deletions(-)
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
index 2104ff42b7b..b0358baebfd 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
@@ -1659,9 +1659,11 @@ public class KafkaConsumerTest {
}
}
+ // NOTE: the CONSUMER protocol path is tested separately in
+ //
FetchRequestManagerTest.testFetchResponseWithUnexpectedPartitionIsIgnored.
@ParameterizedTest
- @EnumSource(GroupProtocol.class)
- public void fetchResponseWithUnexpectedPartitionIsIgnored(GroupProtocol
groupProtocol) {
+ @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
+ public void
testFetchResponseWithUnexpectedPartitionIsIgnored(GroupProtocol groupProtocol) {
ConsumerMetadata metadata = createMetadata(subscription);
MockClient client = new MockClient(time, metadata);
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 8182eee6f16..4a5dafaaeb8 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
@@ -1760,6 +1760,37 @@ public class FetchRequestManagerTest {
fetchedRecords.values().forEach(allFetchedRecords::addAll);
}
+ @Test
+ public void testFetchResponseWithUnexpectedPartitionIsIgnored() {
+ buildFetcher();
+
+ // Only tp0 is assigned and seeked; tp1 is not part of this fetch
session.
+ // When the response includes an unexpected partition (tp1),
FetchSessionHandler
+ // rejects the entire response, so tp0 records are also not returned.
+ assignFromUser(singleton(tp0));
+ subscriptions.seek(tp0, 0);
+
+ assertEquals(1, sendFetches());
+
+ Map<TopicIdPartition, FetchResponseData.PartitionData> partitions =
new LinkedHashMap<>();
+ partitions.put(tidp0, new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp0.partition())
+ .setHighWatermark(100L)
+ .setLogStartOffset(0)
+ .setRecords(records));
+ partitions.put(tidp1, new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp1.partition())
+ .setHighWatermark(100L)
+ .setLogStartOffset(0)
+ .setRecords(records));
+ client.prepareResponse(FetchResponse.of(Errors.NONE, 0,
INVALID_SESSION_ID, new LinkedHashMap<>(partitions), List.of()));
+ networkClientDelegate.poll(time.timer(0));
+
+ assertEquals(0, client.inFlightRequestCount());
+ assertFalse(fetcher.hasCompletedFetches());
+ assertTrue(fetchRecords().isEmpty());
+ }
+
@Test
public void testCompletedFetchRemoval() {
// Ensure the removal of completed fetches that cause an Exception if
and only if they contain empty records.
