bug: Apache Pulsar plugin java.util.ConcurrentModificationException

[Eywek]

Hello,

I'm using the Apache Pulsar stream ingestion plugin (built from commit 0bbb9455eb7aa374e0c8b1e2c5ec0a4d6cd7a93e) and we're seing in the logs the following error:

Mar 29 08:50:57 pinot-hosts-9-server bash[2258505]: 2023/03/29 08:50:57.822 ERROR [LLRealtimeSegmentDataManager_worker_datasource_636bd9305a9600010051fb4d_641de187ab174a000713364a_35__0__8__20230329T0846Z] [worker_datasource_636bd9305a9600010051fb4d_641de187ab174a000713364a_35__0__8__20230329T0846Z] Exception while in work
Mar 29 08:50:57 pinot-hosts-9-server bash[2258505]: java.util.ConcurrentModificationException: null
Mar 29 08:50:57 pinot-hosts-9-server bash[2258505]:         at java.util.ArrayList$Itr.checkForComodification(Unknown Source) ~[?:?]
Mar 29 08:50:57 pinot-hosts-9-server bash[2258505]:         at java.util.ArrayList$Itr.next(Unknown Source) ~[?:?]
Mar 29 08:50:57 pinot-hosts-9-server bash[2258505]:         at org.apache.pinot.shaded.com.google.common.collect.Iterators$6.computeNext(Iterators.java:616) ~[pinot-all-0.13.0-SNAPSHOT-jar-with-dependencies.jar:0.13.0-SNAPSHOT-0bbb9455eb7aa374e0c8b1e2c5ec0a4d6cd7a93e]
Mar 29 08:50:57 pinot-hosts-9-server bash[2258505]:         at org.apache.pinot.shaded.com.google.common.collect.AbstractIterator.tryToComputeNext(AbstractIterator.java:145) ~[pinot-all-0.13.0-SNAPSHOT-jar-with-dependencies.jar:0.13.0-SNAPSHOT-0bbb9455eb7aa374e0c8b1e2c5ec0a4d6cd7a93e]
Mar 29 08:50:57 pinot-hosts-9-server bash[2258505]:         at org.apache.pinot.shaded.com.google.common.collect.AbstractIterator.hasNext(AbstractIterator.java:140) ~[pinot-all-0.13.0-SNAPSHOT-jar-with-dependencies.jar:0.13.0-SNAPSHOT-0bbb9455eb7aa374e0c8b1e2c5ec0a4d6cd7a93e]
Mar 29 08:50:57 pinot-hosts-9-server bash[2258505]:         at java.lang.Iterable.forEach(Unknown Source) ~[?:?]
Mar 29 08:50:57 pinot-hosts-9-server bash[2258505]:         at org.apache.pinot.plugin.stream.pulsar.PulsarMessageBatch.<init>(PulsarMessageBatch.java:50) ~[pinot-pulsar-0.13.0-SNAPSHOT-shaded.jar:0.13.0-SNAPSHOT-68b4096745272706af3e09f08ecbbd89eac7e3a3]
Mar 29 08:50:57 pinot-hosts-9-server bash[2258505]:         at org.apache.pinot.plugin.stream.pulsar.PulsarPartitionLevelConsumer.fetchMessages(PulsarPartitionLevelConsumer.java:89) ~[pinot-pulsar-0.13.0-SNAPSHOT-shaded.jar:0.13.0-SNAPSHOT-68b4096745272706af3e09f08ecbbd89eac7e3a3]
Mar 29 08:50:57 pinot-hosts-9-server bash[2258505]:         at org.apache.pinot.core.data.manager.realtime.LLRealtimeSegmentDataManager.consumeLoop(LLRealtimeSegmentDataManager.java:427) ~[pinot-all-0.13.0-SNAPSHOT-jar-with-dependencies.jar:0.13.0-SNAPSHOT-0bbb9455eb7aa374e0c8b1e2c5ec0a4d6cd7a93e]
Mar 29 08:50:57 pinot-hosts-9-server bash[2258505]:         at org.apache.pinot.core.data.manager.realtime.LLRealtimeSegmentDataManager$PartitionConsumer.run(LLRealtimeSegmentDataManager.java:648) [pinot-all-0.13.0-SNAPSHOT-jar-with-dependencies.jar:0.13.0-SNAPSHOT-0bbb9455eb7aa374e0c8b1e2c5ec0a4d6cd7a93e]
Mar 29 08:50:57 pinot-hosts-9-server bash[2258505]:         at java.lang.Thread.run(Unknown Source) [?:?]

(which seems to stop ingestion btw)

This issue doesn't always happen and is kind of flakky but it's seems to be a fatal one since Pinot stop ingesting any data after this

I wonder if it's not related to messagesList being filtered/used multiple times here

pinot/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarPartitionLevelConsumer.java

Lines 118 to 119 in 0bbb945

	return new PulsarMessageBatch(buildOffsetFilteringIterable(messagesList, startMessageId, endMessageId),
	_enableKeyValueStitch);

and here

pinot/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarPartitionLevelConsumer.java

Lines 89 to 90 in 0bbb945

	return new PulsarMessageBatch(buildOffsetFilteringIterable(messagesList, startMessageId, endMessageId),
	_enableKeyValueStitch);

when the thread is aborted due to timeout or exception

Thank you

[Jackie-Jiang]

cc @KKcorps

[Eywek]

I also have another issue which blocks the fetching:

Mar 30 09:42:37 pinot-hosts-4-server bash[7536]: Exception in thread "worker_datasource_641c1e504381000100a3b7ac_641c35be9a7eee0007e68ae2_3__0__0__20230330T0942Z" java.lang.NoSuchMethodError: 'org.apache.pulsar.client.api.MessageId org.apache.pulsar.client.internal.DefaultImplementation.newMessageId(long, long, int)'
Mar 30 09:42:37 pinot-hosts-4-server bash[7536]:         at org.apache.pinot.plugin.stream.pulsar.PulsarMessageBatch.getNextStreamPartitionMsgOffsetAtIndex(PulsarMessageBatch.java:117)
Mar 30 09:42:37 pinot-hosts-4-server bash[7536]:         at org.apache.pinot.core.data.manager.realtime.LLRealtimeSegmentDataManager.processStreamEvents(LLRealtimeSegmentDataManager.java:613)
Mar 30 09:42:37 pinot-hosts-4-server bash[7536]:         at org.apache.pinot.core.data.manager.realtime.LLRealtimeSegmentDataManager.consumeLoop(LLRealtimeSegmentDataManager.java:450)
Mar 30 09:42:37 pinot-hosts-4-server bash[7536]:         at org.apache.pinot.core.data.manager.realtime.LLRealtimeSegmentDataManager$PartitionConsumer.run(LLRealtimeSegmentDataManager.java:648)
Mar 30 09:42:37 pinot-hosts-4-server bash[7536]:         at java.base/java.lang.Thread.run(Unknown Source)

[Jackie-Jiang]

Solved with #12812