fix(kafka): Call poll() during consumer pause to maintain group membership by graytaylor0 · Pull Request #7072 · opensearch-project/data-prepper

graytaylor0 · 2026-08-04T17:57:39Z

When the circuit breaker or other external condition pauses the Kafka consumer via pauseConsumePredicate, the consume loop previously called Thread.sleep() and then continue, completely skipping the poll() call. Since the Kafka client tracks time between poll() calls via max.poll.interval.ms, exceeding this interval causes the consumer to voluntarily leave the group and trigger a rebalance.

This fix adds a doPoll() call while the consumer is paused. Since consumer.pause() is already called before it, poll() returns zero records while still counting as a valid poll for the interval timer, preventing unnecessary rebalances during backpressure.

Issues Resolved

Resolves #[Issue number to be closed when this PR is merged]

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

…rship When the circuit breaker or other external condition pauses the Kafka consumer via pauseConsumePredicate, the consume loop previously called Thread.sleep() and then continue, completely skipping the poll() call. Since the Kafka client tracks time between poll() calls via max.poll.interval.ms, exceeding this interval causes the consumer to voluntarily leave the group and trigger a rebalance. This fix adds a doPoll() call while the consumer is paused. Since consumer.pause() is already called before it, poll() returns zero records while still counting as a valid poll for the interval timer, preventing unnecessary rebalances during backpressure. Signed-off-by: Taylor Gray <tylgry@amazon.com>

github-actions · 2026-08-04T17:58:47Z

PR Reviewer Guide 🔍

Here are some key observations to aid the review process:

🧪 PR contains tests
🔒 No security concerns identified
✅ No TODO sections
🔀 No multiple PR themes
⚡ Recommended focus areas for review Records dropped during pause `doPoll()` is called while paused, but if records were fetched just before `pause()` took effect (or from partitions that later get unpaused via rebalance), the returned records will be processed and their offsets committed via `iterateRecordPartitions` inside `doPoll()`. More importantly, even with pause, `doPoll()` may return records already buffered from a prior fetch on the just-paused partitions. This bypasses the intended backpressure semantics of `pauseConsuming`, since the whole point of the pause branch was to skip consumption. Consider explicitly ignoring/handling records returned during the pause branch rather than routing them through the normal processing path. doPoll();

github-actions · 2026-08-04T17:59:08Z

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Avoid dropping records returned during pause When pausing due to the external predicate, `doPoll()` may still return records that were prefetched before the pause took effect (Kafka only guarantees the pause takes effect on subsequent fetches). Since the code then `continue`s without processing, these records would be silently dropped. Consider processing the returned records or ensuring the pause is applied before any records are returned. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumer.java [428] } else if (pauseConsumePredicate.pauseConsuming()) { LOG.debug("Pause and skip consuming from Kafka topic due to an external condition: {}", pauseConsumePredicate); paused = true; consumer.pause(consumer.assignment()); - doPoll(); + ConsumerRecords<String, ?> records = doPoll(); + if (records != null && !records.isEmpty()) { + iterateRecordPartitions(records, acknowledgementSet); + } Thread.sleep(1000); continue; Suggestion importance[1-10]: 7 __ Why: Valid concern: `doPoll()` may return prefetched records that would be silently dropped due to the subsequent `continue`. Processing them preserves data integrity.	Medium
General	Replace sleep-based waits with deterministic awaits Relying on `Thread.sleep(2500)` makes this test slow and flaky on loaded CI machines. Consider using an `Awaitility` await block or a `CountDownLatch` triggered from a mock stub to wait deterministically for `resume()` to be invoked before signaling shutdown. data-prepper-plugins/kafka-plugins/src/test/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumerTest.java [918-922] Thread consumerThread = new Thread(() -> consumer.run()); consumerThread.start(); -Thread.sleep(2500); +org.awaitility.Awaitility.await().atMost(Duration.ofSeconds(5)) + .untilAsserted(() -> verify(kafkaConsumer, org.mockito.Mockito.atLeastOnce()).resume(any())); shutdownInProgress.set(true); consumerThread.join(5000); Suggestion importance[1-10]: 5 __ Why: Reasonable improvement to reduce test flakiness and speed up execution, but it's a test quality suggestion with moderate impact.	Low

srikanthpadakanti · 2026-08-13T18:13:31Z

                    LOG.debug("Pause and skip consuming from Kafka topic due to an external condition: {}", pauseConsumePredicate);
                    paused = true;
                    consumer.pause(consumer.assignment());
+                    doPoll();


doPoll() result is intentionally discarded. Add a one-line comment since a rebalance mid-poll could return records that are safely re-delivered later.

srikanthpadakanti · 2026-08-13T18:14:36Z

+        // Need to wait longer than the 1s sleep in the pause branch for the second iteration
+        Thread consumerThread = new Thread(() -> consumer.run());
+        consumerThread.start();
+        Thread.sleep(2500);


Thread.sleep(2500) is slow and timing-dependent. Please use Awaitility.await().untilAsserted(() -> verify(...resume())).

srikanthpadakanti · 2026-08-13T18:15:12Z

+        // Run in a thread and shut down after a short delay
+        Thread consumerThread = new Thread(() -> consumer.run());
+        consumerThread.start();
+        Thread.sleep(100);


Thread.sleep(100) before the verify. I prefer Awaitility over a fixed sleep.

srikanthpadakanti · 2026-08-13T18:16:01Z

+                .thenReturn(true)
+                .thenReturn(false);
+        when(kafkaConsumer.poll(any(Duration.class))).thenReturn(ConsumerRecords.empty());
+        when(kafkaConsumer.assignment()).thenReturn(java.util.Collections.singleton(new TopicPartition(topic, testPartition)));


inline FQN java.util.Collections.singleton(...) Please import Collections.

srikanthpadakanti · 2026-08-13T18:16:46Z

+
+        // Verify poll() was called even though consuming was paused
+        verify(kafkaConsumer, org.mockito.Mockito.atLeastOnce()).pause(any());
+        verify(kafkaConsumer, org.mockito.Mockito.atLeastOnce()).poll(any(Duration.class));


inline FQN org.mockito.Mockito.atLeastOnce() static-import atLeastOnce.

srikanthpadakanti · 2026-08-13T18:16:53Z

+        consumerThread.join(5000);
+
+        // Verify poll() was called even though consuming was paused
+        verify(kafkaConsumer, org.mockito.Mockito.atLeastOnce()).pause(any());


inline FQN org.mockito.Mockito.atLeastOnce() static-import atLeastOnce.

srikanthpadakanti · 2026-08-13T18:17:05Z

+        consumerThread.join(5000);
+
+        // Verify that consumer was paused and then resumed
+        verify(kafkaConsumer, org.mockito.Mockito.atLeastOnce()).pause(any());


inline FQN org.mockito.Mockito.atLeastOnce() static-import atLeastOnce.

srikanthpadakanti · 2026-08-13T18:17:13Z

+
+        // Verify that consumer was paused and then resumed
+        verify(kafkaConsumer, org.mockito.Mockito.atLeastOnce()).pause(any());
+        verify(kafkaConsumer, org.mockito.Mockito.atLeastOnce()).resume(any());


inline FQN org.mockito.Mockito.atLeastOnce() static-import atLeastOnce.

srikanthpadakanti

Hey @graytaylor0 Thank you for your contribution. Please address the comments.

graytaylor0 requested review from KarstenSchnitter, Zhangxunmt, dinujoh, divbok, dlvenable, kkondaka, oeyh, san81, sb2k16, srikanthjg and srikanthpadakanti as code owners August 4, 2026 17:57

srikanthpadakanti reviewed Aug 13, 2026

View reviewed changes

srikanthpadakanti requested changes Aug 13, 2026

View reviewed changes

Navigation Menu

Uh oh!

fix(kafka): Call poll() during consumer pause to maintain group membership - #7072

fix(kafka): Call poll() during consumer pause to maintain group membership#7072
graytaylor0 wants to merge 1 commit into
opensearch-project:mainfrom
graytaylor0:KafkaPoll

graytaylor0 commented Aug 4, 2026 •

edited

Loading

Uh oh!

github-actions Bot commented Aug 4, 2026

Uh oh!

github-actions Bot commented Aug 4, 2026

Uh oh!

srikanthpadakanti Aug 13, 2026

Uh oh!

srikanthpadakanti Aug 13, 2026

Uh oh!

srikanthpadakanti Aug 13, 2026

Uh oh!

srikanthpadakanti Aug 13, 2026

Uh oh!

srikanthpadakanti Aug 13, 2026

Uh oh!

srikanthpadakanti Aug 13, 2026

Uh oh!

srikanthpadakanti Aug 13, 2026

Uh oh!

srikanthpadakanti Aug 13, 2026

Uh oh!

srikanthpadakanti left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

graytaylor0 commented Aug 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Issues Resolved

Check List

Uh oh!

github-actions Bot commented Aug 4, 2026

PR Reviewer Guide 🔍

Uh oh!

github-actions Bot commented Aug 4, 2026

PR Code Suggestions ✨

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

srikanthpadakanti left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

graytaylor0 commented Aug 4, 2026 •

edited

Loading