You are viewing a plain text version of this content. The canonical link for it is here.

Posted to jira@kafka.apache.org by "kirktrue (via GitHub)" <gi...@apache.org> on 2025/06/17 13:42:49 UTC

[PR] KAFKA-19259: Async consumer fetch intermittent delays on console consumer [kafka]

kirktrue opened a new pull request, #19980:
URL: https://github.com/apache/kafka/pull/19980

   There’s a difference in the two consumers’ `pollForFetches()` methods in this case: `ClassicKafkaConsumer` doesn't block waiting for data in the fetch buffer, but `AsyncKafkaConsumer` does.
   
   In `ClassicKafkaConsumer.pollForFetches()`, after enqueuing the `FETCH` request, the consumer makes a call to `ConsumerNetworkClient.poll()`. In most cases `poll()` returns almost immediately because it successfully sent the `FETCH` request. So even when the `pollTimeout` value is, e.g. 3000, the call to `ConsumerNetworkClient.poll()` doesn't block that long waiting for a response.
   
   After sending out a `FETCH` request, `AsyncKafkaConsumer` then calls `FetchBuffer.awaitNotEmpty()` and proceeds to block there for the full length of the timeout. In some cases, the response to the `FETCH` comes back with no results, which doesn't unblock `FetchBuffer.awaitNotEmpty()`. So because the application thread is still waiting for data in the buffer, it remains blocked, preventing any more `FETCH` requests from being sent, causing the long pauses in the console consumer.


-- 
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-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-19259: Async consumer fetch intermittent delays on console consumer [kafka]

Posted by "lianetm (via GitHub)" <gi...@apache.org>.

lianetm commented on code in PR #19980:
URL: https://github.com/apache/kafka/pull/19980#discussion_r2316339959


##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerTest.java:
##########
@@ -1588,6 +1590,75 @@ private void sendCompressedMessages(int numRecords, TopicPartition tp) {
         }
     }
 
+    @ClusterTest
+    public void testClassicConsumerStallBetweenPoll() throws Exception {
+        testStallBetweenPoll(GroupProtocol.CLASSIC);
+    }
+
+    @ClusterTest
+    public void testAsyncConsumerStallBetweenPoll() throws Exception {
+        testStallBetweenPoll(GroupProtocol.CONSUMER);
+    }
+
+    /**
+     * This test is to prove that the intermittent stalling that has been experienced when using the asynchronous
+     * consumer, as filed under KAFKA-19259, have been fixed.
+     *
+     * <p/>
+     *
+     * The basic idea is to have one thread that produces a record every 500 ms. and the main thread that consumes
+     * records without pausing between polls for much more than the produce delay. In the test case filed in
+     * KAFKA-19259, the consumer sometimes pauses for up to 5-10 seconds despite records being produced every second.
+     */
+    private void testStallBetweenPoll(GroupProtocol groupProtocol) throws Exception {
+        var testTopic = "stall-test-topic";
+        var numPartitions = 6;
+        cluster.createTopic(testTopic, numPartitions, (short) BROKER_COUNT);
+
+        // The producer must produce slowly to tickle the scenario.
+        var produceDelay = 500;
+
+        var executor = Executors.newScheduledThreadPool(1);
+
+        try (var producer = cluster.producer()) {
+            // Start a thread running that produces records at a relative trickle.
+            executor.scheduleWithFixedDelay(
+                () -> producer.send(new ProducerRecord<>(testTopic, TestUtils.randomBytes(64))),
+                0,
+                produceDelay,
+                TimeUnit.MILLISECONDS
+            );
+
+            Map<String, Object> consumerConfig = Map.of(GROUP_PROTOCOL_CONFIG, groupProtocol.name().toLowerCase(Locale.ROOT));
+
+            // Assign a tolerance for how much time is allowed to pass between Consumer.poll() calls given that there
+            // should be *at least* one record to read every second.
+            var pollDelayTolerance = 2000;
+
+            try (Consumer<byte[], byte[]> consumer = cluster.consumer(consumerConfig)) {
+                consumer.subscribe(List.of(testTopic));
+
+                // This is here to allow the consumer time to settle the group membership/assignment.
+                awaitNonEmptyRecords(consumer, new TopicPartition(testTopic, 0));
+
+                // Keep track of the last time the poll is invoked to ensure the deltas between invocations don't
+                // exceed the delay threshold defined above.
+                var beforePoll = System.currentTimeMillis();
+                consumer.poll(Duration.ofSeconds(5));
+                consumer.poll(Duration.ofSeconds(5));
+                var afterPoll = System.currentTimeMillis();
+                var pollDelay = afterPoll - beforePoll;
+
+                if (pollDelay > pollDelayTolerance)
+                    fail("Detected a stall of " + pollDelay + " ms between Consumer.poll() invocations despite a Producer producing records every " + produceDelay + " ms");

Review Comment:
   Adding a `consumer.unsubscribe()` before completing the test should help solve the flakiness @AndrewJSchofield reported.
   
   I tried locally and got it too (around 30 iterations), and noticed it fails on the first poll to join the group and consume. This seemed to solve the flakiness on my side, up to 100 iterations, but please double check in case there is something else.



##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchBuffer.java:
##########
@@ -98,7 +98,8 @@ void add(CompletedFetch completedFetch) {
         try {
             lock.lock();
             completedFetches.add(completedFetch);
-            notEmptyCondition.signalAll();
+            wokenup.set(true);
+            blockingCondition.signalAll();

Review Comment:
   could we reuse the existing `addAll` in this implementation of `add`? Would avoid dup logic like this



-- 
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-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org