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 0071092b36b KAFKA-16630: Fix flaky classic consumer poll test (#22787)
0071092b36b is described below
commit 0071092b36bf2f090d725643cfe0e69101d3f474
Author: Lianet Magrans <[email protected]>
AuthorDate: Thu Jul 9 09:30:39 2026 -0400
KAFKA-16630: Fix flaky classic consumer poll test (#22787)
Flakiness reproduced locally, due to a single call to poll(0) that may
not be enough if the HB thread interleaves with the application thread.
In the case where HB thread removes the completion handler for the fetch
request from the queue, but before firing it (what puts the records on
the buffer), the app thread reads interleaves, no completed request
found in the queue (so it returns empty on its single poll(0) attempt)
Ensuring there is a following poll will allow to see the records (even
if it was the HB thread taking the completed request from the queue, it
will just fire completion and the next poll after that finds the records
in the buffer).
Same fix applied to 2 tests that had the same check.
Reviewers: Andrew Schofield <[email protected]>, TengYao Chi
<[email protected]>
---
.../kafka/clients/consumer/KafkaConsumerTest.java | 27 +++++++++++++++++-----
1 file changed, 21 insertions(+), 6 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 aa596cf866c..c26e5eb0c29 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
@@ -491,11 +491,19 @@ public class KafkaConsumerTest {
@ParameterizedTest
@EnumSource(value = GroupProtocol.class, names = "CLASSIC")
@SuppressWarnings("unchecked")
- public void testPollReturnsRecords(GroupProtocol groupProtocol) {
+ public void testPollReturnsRecords(GroupProtocol groupProtocol) throws
InterruptedException {
consumer = setUpConsumerWithRecordsToPoll(groupProtocol, tp0, 5);
- ConsumerRecords<String, String> records = (ConsumerRecords<String,
String>) consumer.poll(Duration.ZERO);
+ // Poll until the expected records are consumed. A single
poll(Duration.ZERO) may not be enough
+ // and make the test flaky, as it can return empty when the heartbeat
thread and the consumer
+ // thread race to fire the fetch completion.
+ AtomicReference<ConsumerRecords<String, String>> polled = new
AtomicReference<>(ConsumerRecords.empty());
+ TestUtils.waitForCondition(() -> {
+ polled.set((ConsumerRecords<String, String>)
consumer.poll(Duration.ZERO));
+ return polled.get().count() == 5;
+ }, "Consumer did not return the fetched records.");
+ ConsumerRecords<String, String> records = polled.get();
assertEquals(5, records.count());
assertEquals(Set.of(tp0), records.partitions());
assertEquals(5, records.records(tp0).size());
@@ -507,14 +515,23 @@ public class KafkaConsumerTest {
@ParameterizedTest
@EnumSource(value = GroupProtocol.class, names = "CLASSIC")
@SuppressWarnings("unchecked")
- public void
testSecondPollWithDeserializationErrorThrowsRecordDeserializationException(GroupProtocol
groupProtocol) {
+ public void
testSecondPollWithDeserializationErrorThrowsRecordDeserializationException(GroupProtocol
groupProtocol) throws InterruptedException {
int invalidRecordNumber = 4;
int invalidRecordOffset = 3;
StringDeserializer deserializer =
mockErrorDeserializer(invalidRecordNumber);
consumer = setUpConsumerWithRecordsToPoll(groupProtocol, tp0, 5,
deserializer);
- ConsumerRecords<String, String> records = (ConsumerRecords<String,
String>) consumer.poll(Duration.ZERO);
+ // Poll until the expected records are consumed. A single
poll(Duration.ZERO) may not be enough
+ // and make the test flaky, as it can return empty when the heartbeat
thread and the consumer
+ // thread race to fire the fetch completion.
+ AtomicReference<ConsumerRecords<String, String>> polled = new
AtomicReference<>(ConsumerRecords.empty());
+ TestUtils.waitForCondition(() -> {
+ polled.set((ConsumerRecords<String, String>)
consumer.poll(Duration.ZERO));
+ return polled.get().count() == invalidRecordNumber - 1;
+ }, "Consumer did not return the records preceding the deserialization
error.");
+
+ ConsumerRecords<String, String> records = polled.get();
assertEquals(invalidRecordNumber - 1, records.count());
assertEquals(Set.of(tp0), records.partitions());
assertEquals(invalidRecordNumber - 1, records.records(tp0).size());
@@ -2370,8 +2387,6 @@ public class KafkaConsumerTest {
consumer2.close(CloseOptions.timeout(Duration.ZERO));
}
- // TODO: this test references RPCs to be sent that are not part of the
CONSUMER group protocol.
- // We are deferring any attempts at generalizing this test for both
group protocols to the future.
@ParameterizedTest
@EnumSource(value = GroupProtocol.class, names = "CLASSIC")
@SuppressWarnings("unchecked")
