Wie kann ich ConcurrentModificationException Fehler in Kafka beheben? (0.9.0.1)

Question

Mir ist Kafkas Aussage bekannt, dass KafkaConsumer nicht Thread-sicher ist.

Also tat ich dies: (Scala)

val m = Map(new TopicPartition(msg.topic(), msg.partition()) -> new OffsetAndMetadata(msg.offset())) 
consumer.synchronized{ consumer.commitSync(m) } 

ich meinen Zugang zum Verbraucher in einem synchronisierten Block setzen, aber ich habe noch ConcurrentModificationException Fehler auf der Linie mit consumer.commitSync (m) erhalten.

Warum und was kann ich dagegen tun?

Ich benutze Akka-Streams, also gibt es bestimmt Threads-Geheimnisse unter der Decke, aber sollte nicht der synchronisierte Block dafür sorgen?

Answer 1

Ein Ansatz in der Dokumentation besteht darin, einen separaten Thread nur für den KafkaConsumer zu erstellen und mit der Außenwelt mit einer Art gleichzeitiger Warteschlange zu kommunizieren.