(camel) branch main updated: CAMEL-22004: camel-kafka - Pause action must be re-issued when new partitions are assigned to preserve the paused state of the Consumer assignment (#17831)

[da...@apache.org]

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

davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git


The following commit(s) were added to refs/heads/main by this push:
     new 9230868edcc CAMEL-22004: camel-kafka - Pause action must be re-issued when new partitions are assigned to preserve the paused state of the Consumer assignment (#17831)
9230868edcc is described below

commit 9230868edcc79eb3a6b36b7f61cb769463616502
Author: Pavel Bořík <32...@users.noreply.github.com>
AuthorDate: Tue Apr 22 16:43:51 2025 +0200

    CAMEL-22004: camel-kafka - Pause action must be re-issued when new partitions are assigned to preserve the paused state of the Consumer assignment (#17831)
---
 .../camel/component/kafka/KafkaFetchRecords.java   | 48 +++++++++++++++++-----
 1 file changed, 38 insertions(+), 10 deletions(-)

diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
index 99e4d1157c3..83058a95861 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
@@ -17,8 +17,10 @@
 package org.apache.camel.component.kafka;
 
 import java.time.Duration;
+import java.util.Collection;
 import java.util.Properties;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.regex.Pattern;
 
@@ -96,7 +98,7 @@ public class KafkaFetchRecords implements Runnable {
     private volatile long currentBackoffInterval;
     private volatile boolean reconnect; // The reconnect must be false at init (this is the policy whether to reconnect).
     private volatile boolean connected; // this is the state (connected or not)
-    private volatile State state = State.RUNNING;
+    private final AtomicReference<State> state = new AtomicReference<>(State.RUNNING);
 
     private final DevConsoleMetricsCollector metricsCollector;
 
@@ -337,7 +339,7 @@ public class KafkaFetchRecords implements Runnable {
         LOG.info("Searching for a custom subscribe adapter on the registry");
         final SubscribeAdapter adapter = resolveSubscribeAdapter(camelContext);
 
-        adapter.subscribe(consumer, listener, topicInfo);
+        adapter.subscribe(consumer, new PausePreservingRebalanceListener(listener), topicInfo);
     }
 
     private SubscribeAdapter resolveSubscribeAdapter(CamelContext camelContext) {
@@ -388,7 +390,7 @@ public class KafkaFetchRecords implements Runnable {
                 updateTaskState();
 
                 // when breakOnFirstError we want to unsubscribe from Kafka
-                if (result != null && result.isBreakOnErrorHit() && !this.state.equals(State.PAUSED)) {
+                if (result != null && result.isBreakOnErrorHit() && !this.state.get().equals(State.PAUSED)) {
                     LOG.debug("We hit an error ... setting flags to force reconnect");
                     // force re-connect
                     setReconnect(true);
@@ -452,11 +454,11 @@ public class KafkaFetchRecords implements Runnable {
     }
 
     private void updateTaskState() {
-        switch (state) {
+        switch (state.get()) {
             case PAUSE_REQUESTED:
                 LOG.info("Pausing the consumer as a response to a pause request");
                 consumer.pause(consumer.assignment());
-                state = State.PAUSED;
+                state.set(State.PAUSED);
                 break;
             case RESUME_REQUESTED:
                 LOG.info("Resuming the consumer as a response to a resume request");
@@ -471,7 +473,7 @@ public class KafkaFetchRecords implements Runnable {
                     });
                 }
                 consumer.resume(consumer.assignment());
-                state = State.RUNNING;
+                state.set(State.RUNNING);
                 break;
             default:
                 break;
@@ -585,7 +587,7 @@ public class KafkaFetchRecords implements Runnable {
     public boolean isPaused() {
         // cannot use consumer directly as you can have ConcurrentModificationException as kafka client does not permit
         // multiple threads to use the client consumer, so we check the state only
-        return state == State.PAUSED;
+        return state.get() == State.PAUSED;
     }
 
     public void setConnected(boolean connected) {
@@ -652,7 +654,7 @@ public class KafkaFetchRecords implements Runnable {
      */
     public void pause() {
         LOG.info("A pause request was issued and the consumer thread will pause after current processing has finished");
-        state = State.PAUSE_REQUESTED;
+        state.set(State.PAUSE_REQUESTED);
     }
 
     /*
@@ -661,7 +663,7 @@ public class KafkaFetchRecords implements Runnable {
      */
     public void resume() {
         LOG.info("A resume request was issued and the consumer thread will resume after current processing has finished");
-        state = State.RESUME_REQUESTED;
+        state.set(State.RESUME_REQUESTED);
     }
 
     private void setLastError(Exception lastError) {
@@ -677,6 +679,32 @@ public class KafkaFetchRecords implements Runnable {
     }
 
     String getState() {
-        return state.name();
+        return state.get().name();
+    }
+
+    private class PausePreservingRebalanceListener implements ConsumerRebalanceListener {
+        private final ConsumerRebalanceListener delegate;
+
+        PausePreservingRebalanceListener(ConsumerRebalanceListener delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
+            delegate.onPartitionsRevoked(partitions);
+        }
+
+        @Override
+        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+            if (state.compareAndSet(State.PAUSED, State.PAUSE_REQUESTED)) {
+                LOG.debug("Partitions were assigned while paused, the consumer will be re-paused");
+            }
+            delegate.onPartitionsAssigned(partitions);
+        }
+
+        @Override
+        public void onPartitionsLost(Collection<TopicPartition> partitions) {
+            delegate.onPartitionsLost(partitions);
+        }
     }
 }