Repository navigation
ConcurrentModificationException in ShardManager during Rebalancing #188
Description
Hi,
we're testing a pre-version of the Parallel Consumer from the features/batching branch. I however suspect that the issue we see is not related to the changes on that branch.
Setup:
We have two topics with multiple partitions each and we have two consumer groups with multiple consumers in each group which are starting up pretty much in parallel.
Problem:
Sometimes during startup we see an exception like this, which seems to be related to a race condition when accessing the ShardManager while a rebalancing is happening:
[INFO ] 2022-02-07 16:30:43.919 [pc-pool-2-thread-1] o.a.k.c.u.AppInfoParser - Kafka version: 2.6.0
[INFO ] 2022-02-07 16:30:43.919 [pc-pool-2-thread-1] o.a.k.c.u.AppInfoParser - Kafka commitId: 62abe01bee039651
[INFO ] 2022-02-07 16:30:43.919 [pc-pool-2-thread-1] o.a.k.c.u.AppInfoParser - Kafka startTimeMs: 1644247843892
[INFO ] 2022-02-07 16:30:43.993 [kafka-producer-network-thread | producer-1] o.a.k.c.Metadata - [Producer clientId=producer-1] Cluster ID: jfbZl7cxTwmF0aBasWKl1A
[INFO ] 2022-02-07 16:30:44.611 [pc-broker-poll] o.a.k.c.c.i.AbstractCoordinator - [Consumer clientId=consumer-test-consumer-group2, groupId=test-consumer-group] Attempt to heartbeat failed since group is rebalancing
[INFO ] 2022-02-07 16:30:44.612 [pc-broker-poll] o.a.k.c.c.i.ConsumerCoordinator - [Consumer clientId=consumer-test-consumer-group2, groupId=test-consumer-group] Revoke previously assigned partitions test-topic-4, test-topic-5, test-topic-6, test-topic-7
[INFO ] 2022-02-07 16:30:44.612 [pc-broker-poll] i.c.p.s.PartitionMonitor - Partitions revoked: [test-topic-4, test-topic-5, test-topic-6, test-topic-7]
[INFO ] 2022-02-07 16:30:44.615 [pc-broker-poll] o.a.k.c.c.i.AbstractCoordinator - [Consumer clientId=consumer-test-consumer-group1, groupId=test-consumer-group] Attempt to heartbeat failed since group is rebalancing
[INFO ] 2022-02-07 16:30:44.615 [pc-broker-poll] o.a.k.c.c.i.AbstractCoordinator - [Consumer clientId=consumer-test-consumer-group2, groupId=test-consumer-group] (Re-)joining group
[INFO ] 2022-02-07 16:30:44.615 [pc-broker-poll] o.a.k.c.c.i.ConsumerCoordinator - [Consumer clientId=consumer-test-consumer-group1, groupId=test-consumer-group] Revoke previously assigned partitions other-test-topic-12, other-test-topic-11, other-test-topic-14, other-test-topic-13, other-test-topic-10
[INFO ] 2022-02-07 16:30:44.617 [pc-broker-poll] i.c.p.s.PartitionMonitor - Partitions revoked: [other-test-topic-12, other-test-topic-11, other-test-topic-14, other-test-topic-13, other-test-topic-10]
[INFO ] 2022-02-07 16:30:44.622 [pc-broker-poll] o.a.k.c.c.i.AbstractCoordinator - [Consumer clientId=consumer-test-consumer-group1, groupId=test-consumer-group] (Re-)joining group
[ERROR] 2022-02-07 16:30:44.617 [pc-control] i.c.p.i.AbstractParallelEoSStreamProcessor - Error from poll control thread, will attempt controlled shutdown, then rethrow. Error: null
java.util.ConcurrentModificationException: null
at java.util.HashMap$HashIterator.nextNode(HashMap.java:1445) ~[?:1.8.0_292]
at java.util.HashMap$EntryIterator.next(HashMap.java:1479) ~[?:1.8.0_292]
at java.util.HashMap$EntryIterator.next(HashMap.java:1477) ~[?:1.8.0_292]
at io.confluent.parallelconsumer.state.ShardManager.getWorkQueuedInShardsCount(ShardManager.java:73) ~[parallel-consumer-core-0.4.0.2-batching-2.jar!/:?]
at io.confluent.parallelconsumer.state.WorkManager.getTotalWorkWaitingProcessing(WorkManager.java:595) ~[parallel-consumer-core-0.4.0.2-batching-2.jar!/:?]
at io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor.controlLoop(AbstractParallelEoSStreamProcessor.java:652) ~[parallel-consumer-core-0.4.0.2-batching-2.jar!/:?]
at io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor.lambda$supervisorLoop$5(AbstractParallelEoSStreamProcessor.java:565) ~[parallel-consumer-core-0.4.0.2-batching-2.jar!/:?]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_292]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_292]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_292]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_292]
Note: We have two instances of the ParallelConsumer (with their own KafkaConsumer in the background) as the processing logic from the two topics is slightly different. From the log it's not really clear which message is coming from which ParallelConsumer instance as the thread names are not uniquely generated.
Cheers,
Niels
[EDIT] Added more details
Activity
astubbs commentedon Feb 11, 2022
Thanks for the issue submission.
I think I see the issue and have resolved it..
Do you mean multiple PC's?
Out of curiosity, how many?
astubbs commentedon Feb 11, 2022
(And yes, you're correct - unrelated to the batching patch)
nioertel commentedon Feb 11, 2022
Hi Anthony,
within the specific application we have 4 PC instances (with their own KafkaConsumer instances behind). They have lots of shared code but process from different topics with different internal configurations of the entry points. That's the reason why we decided to include them in once application.
So it would be nice if you have some static AtomicInteger somewhere that is used to generate an id which is included in all thread names so the logs can be directly related to a specific PC instance.
However we also have multiple consumers in each consumer group (i.e. application processes of the same consumer group on the same topic) for high availability and load balancing across multiple physical machines. That's actually the cause for the quite busy rebalancing during startup:
astubbs commentedon Feb 11, 2022
So how many PC instances in total?
And yes we can add some option for log prefixing - can you create a separate issue for that please?
nioertel commentedon Feb 11, 2022
Our current setup is 8 PC instances in 8 separate Java processes with the same consumer group processing from the same topic which has 24 partitions.
Each Java process however is processing from 4 topics with separate PC instances for each so the total number of PC instances is 32.
astubbs commentedon Feb 11, 2022
So to be clear, is that 8*8 pc instances that have the same subscription - so 64 in total? Wait - now you got me confused again :)
astubbs commentedon Feb 11, 2022
Sounds like 8 * 8 with the same group and sub subscription plus 4 sets of 8 with same group and subscription.
So the highest count which share a group and subscription is the 8 * 8 set?
And the total count of pc instances (with same group or not) is 8x8 plus 4x8?
nioertel commentedon Feb 11, 2022
The highest count that share a subscription is 8 (all in separate Java processes).
astubbs commentedon Feb 11, 2022
Haha ok thanks :) I'll write the test with significantly more anyway :)
astubbs commentedon Feb 15, 2022
FYI Patch available here: #190
astubbs commentedon Feb 24, 2022
Relates:
fix: Concurrency and State improvements (#190)
fix: Concurrency and State improvements (#190)
fix: Concurrency and State improvements (#190)
docs: qualify every issue reference tree-wide
docs: qualify every issue reference tree-wide