[CAMEL-17285] camel-kafka: async manual commit is not working

• Type: Test
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 3.14.0
• Fix Version/s: 3.14.0
• Component/s: camel-kafka
• Estimated Complexity: Unknown
• Assignee: Otavio Rodolfo Piske
• Reporter: Otavio Rodolfo Piske
• Created: 06/Dec/21 16:19
• Updated: 06/Dec/21 20:19
• Resolved: 06/Dec/21 19:25

Description

Recently the async manual commit support for Kafka has stopped working. It needs to be investigated.

 

Kafka's client is not thread safe and a recent change may have moved the thread responsible for creating it and/or performing the commit:

 

 
java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
        at org.apache.kafka.clients.consumer.KafkaConsumer.acquire(KafkaConsumer.java:2445) ~[kafka-clients-2.8.1.jar:?]
        at org.apache.kafka.clients.consumer.KafkaConsumer.acquireAndEnsureOpen(KafkaConsumer.java:2429) ~[kafka-clients-2.8.1.jar:?]
        at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1486) ~[kafka-clients-2.8.1.jar:?]
        at org.apache.camel.component.kafka.DefaultKafkaManualSyncCommit.commitOffset(DefaultKafkaManualSyncCommit.java:58) ~[classes/:?]
        at org.apache.camel.component.kafka.DefaultKafkaManualSyncCommit.commit(DefaultKafkaManualSyncCommit.java:49) ~[classes/:?]
        at org.apache.camel.component.kafka.integration.KafkaConsumerAsyncManualCommitIT$1.lambda$configure$0(KafkaConsumerAsyncManualCommitIT.java:92) ~[test-classes/:?]
        at org.apache.camel.support.processor.DelegateSyncProcessor.process(DelegateSyncProcessor.java:65) ~[classes/:?]
        at org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$SimpleTask.run(RedeliveryErrorHandler.java:469) ~[classes/:?]
        at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:187) ~[classes/:?]
        at org.apache.camel.impl.engine.DefaultReactiveExecutor.schedule(DefaultReactiveExecutor.java:59) ~[classes/:?]
        at org.apache.camel.processor.aggregate.AggregateProcessor.lambda$onSubmitCompletion$4(AggregateProcessor.java:891) ~[classes/:?]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
        at java.lang.Thread.run(Thread.java:829) [?:?]

 

Issue Links

• links to: GitHub Pull Request #6507