[CAMEL-18003] camel-kafka: manual pause of consumer causes ConcurrentModificationException

• Type: Task
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Component/s: camel-health, camel-kafka
• Estimated Complexity: Unknown
• Assignee: Otavio Rodolfo Piske
• Reporter: Otavio Rodolfo Piske
• Created: 22/Apr/22 09:42
• Updated: 29/Apr/22 10:21
• Resolved: 29/Apr/22 10:21

Description

The Kafka consumer is not safe for concurrent access. We need to protect it from multiple different threads trying to access the instance on the component consumer. Otherwise:

java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
        at org.apache.kafka.clients.consumer.KafkaConsumer.acquire(KafkaConsumer.java:2469) ~[kafka-clients-3.1.0.jar:?]
        at org.apache.kafka.clients.consumer.KafkaConsumer.acquireAndEnsureOpen(KafkaConsumer.java:2453) ~[kafka-clients-3.1.0.jar:?]
        at org.apache.kafka.clients.consumer.KafkaConsumer.assignment(KafkaConsumer.java:891) ~[kafka-clients-3.1.0.jar:?]
        at org.apache.camel.component.kafka.KafkaFetchRecords.pause(KafkaFetchRecords.java:519) ~[camel-kafka-3.17.0-SNAPSHOT.jar:3.17.0-SNAPSHOT]
        at org.apache.camel.component.kafka.KafkaConsumer.doSuspend(KafkaConsumer.java:221) ~[camel-kafka-3.17.0-SNAPSHOT.jar:3.17.0-SNAPSHOT]
        at org.apache.camel.support.service.BaseService.suspend(BaseService.java:189) ~[camel-api-3.17.0-SNAPSHOT.jar:3.17.0-SNAPSHOT]
        at org.apache.camel.support.service.ServiceHelper.suspendService(ServiceHelper.java:404) ~[camel-api-3.17.0-SNAPSHOT.jar:3.17.0-SNAPSHOT]
        at org.apache.camel.impl.engine.DefaultShutdownStrategy.suspendNow(DefaultShutdownStrategy.java:447) ~[camel-base-engine-3.17.0-SNAPSHOT.jar:3.17.0-SNAPSHOT]
        at org.apache.camel.impl.engine.DefaultShutdownStrategy$ShutdownTask.run(DefaultShutdownStrategy.java:620) ~[camel-base-engine-3.17.0-SNAPSHOT.jar:3.17.0-SNAPSHOT]
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) ~[?:?]
        at java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]
        at java.lang.Thread.run(Thread.java:833) ~[?:?]

Issue Links

• is related to: CAMEL-17051 camel-kafka: Need the ability to pause kafka consumer
• relates to: CAMEL-18019 camel-kafka: review fields for concurrent access
• links to: GitHub Pull Request #7502