[camel] branch main updated: CAMEL-18003: fix incorrect concurrent access to the Kafka consumer

[or...@apache.org]

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

orpiske 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 2f589ad130d CAMEL-18003: fix incorrect concurrent access to the Kafka consumer
2f589ad130d is described below

commit 2f589ad130daacd292fdbb7ef2ef5febf271cf90
Author: Otavio Rodolfo Piske <an...@gmail.com>
AuthorDate: Wed Apr 27 12:53:33 2022 +0200

    CAMEL-18003: fix incorrect concurrent access to the Kafka consumer
    
    This prevents the code from throwing ConcurrentModificationException
    when suspending
---
 .../camel/component/kafka/KafkaFetchRecords.java   | 39 ++++++++++++++++++++--
 1 file changed, 37 insertions(+), 2 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 0c54ea23363..1e5a4be5403 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
@@ -50,6 +50,19 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class KafkaFetchRecords implements Runnable {
+    /*
+     This keeps track of the state the record fetcher is. Because the Kafka consumer is not thread safe, it may take
+     some time between the pause or resume request is triggered and it is actually set.
+     Some of the states may be set but not read. This is done deliberately to avoid the code to enter multiple times
+     the branches that handle the *_REQUESTED states.
+     */
+    private enum State {
+        RUNNING,
+        PAUSE_REQUESTED,
+        PAUSED,
+        RESUME_REQUESTED,
+    }
+
     private static final Logger LOG = LoggerFactory.getLogger(KafkaFetchRecords.class);
 
     private final KafkaConsumer kafkaConsumer;
@@ -72,6 +85,8 @@ public class KafkaFetchRecords implements Runnable {
     private boolean reconnect; // must be false at init (this is the policy whether to reconnect)
     private boolean connected; // this is the state (connected or not)
 
+    private volatile State state = State.RUNNING;
+
     KafkaFetchRecords(KafkaConsumer kafkaConsumer,
                       BridgeExceptionHandlerToErrorHandler bridge, String topicName, Pattern topicPattern, String id,
                       Properties kafkaProps, KafkaConsumerListener consumerListener) {
@@ -313,6 +328,7 @@ public class KafkaFetchRecords implements Runnable {
                     setConnected(false);
                 }
 
+                updateTaskState();
             }
 
             if (!isConnected()) {
@@ -359,6 +375,23 @@ public class KafkaFetchRecords implements Runnable {
         }
     }
 
+    private void updateTaskState() {
+        switch (state) {
+            case PAUSE_REQUESTED:
+                LOG.info("Pausing the consumer as a response to a pause request");
+                consumer.pause(consumer.assignment());
+                state = State.PAUSED;
+                break;
+            case RESUME_REQUESTED:
+                LOG.info("Resuming the consumer as a response to a resume request");
+                consumer.resume(consumer.assignment());
+                state = State.RUNNING;
+                break;
+            default:
+                break;
+        }
+    }
+
     private void safeUnsubscribe() {
         if (consumer == null) {
             return;
@@ -516,7 +549,8 @@ public class KafkaFetchRecords implements Runnable {
      * or via JMX
      */
     public void pause() {
-        consumer.pause(consumer.assignment());
+        LOG.info("A pause request was issued and the consumer thread will pause after current processing has finished");
+        state = State.PAUSE_REQUESTED;
     }
 
     /*
@@ -524,6 +558,7 @@ public class KafkaFetchRecords implements Runnable {
      * mostly used for directly calling resume from Java code or via JMX
      */
     public void resume() {
-        consumer.resume(consumer.assignment());
+        LOG.info("A resume request was issued and the consumer thread will resume after current processing has finished");
+        state = State.RESUME_REQUESTED;
     }
 }