confluentinc#319: Error during shutdown - ConcurrentModificationException #252

Mirror of confluentinc/parallel-consumer#319
Upstream title (verbatim, 2023-07-07): Error during shutdown - ConcurrentModificationException
Opened by JorgenRingen, 2022-06-02 · no upstream reply before closure
Closed upstream 2023-07-07 as completed by an administrative sweep - but the bug genuinely is
fixed, for reasons the sweep did not know. See below.
Summarised, not copied. Discussion belongs here - the Confluent-hosted repository is no longer maintained.

Summary

ConcurrentModificationException during close, seen once on 0.4.0.1. Reported as a courtesy - the
reporter had been told it should be fixed by upstream PR #270 and posted the trace in case others
hit it.

java.util.ConcurrentModificationException: null
  at java.util.HashMap$ValueSpliterator.forEachRemaining(HashMap.java:1784)
  at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
  ...
  at io.confluent.parallelconsumer.state.PartitionMonitor.getNumberOfEntriesInPartitionQueues(PartitionMonitor.java:258)
  at io.confluent.parallelconsumer.state.WorkManager.getNumberOfEntriesInPartitionQueues(WorkManager.java:412)
  at io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor.controlLoop(AbstractParallelEoSStreamProcessor.java:650)

The control loop streamed the values of a plain HashMap of partition state while another thread
mutated it.

Fork status

Fixed - verified by reading the code. Closing as genuinely done.

Recorded rather than simply dropped, because the upstream close was administrative and therefore not
evidence of anything: the fix had to be confirmed independently. Note also that PR #270, the fix the
reporter was promised, was itself closed unmerged in the 2023-06-15 PR sweep - so the bug was
fixed by a different route than anyone in the thread expected.

Three things changed, and each alone would prevent this trace:

1. PartitionMonitor was renamed to PartitionStateManager (upstream feat(connect-spike) astubbs#240: Connect sinks on PC - direction, offset analysis, and the shadowing proof (confluentinc#119) #269), and
   getNumberOfEntriesInPartitionQueues no longer exists - the control loop now calls
   getNumberOfWorkQueuedInShardsAwaitingSelection() and getNumberOfIncompleteOffsets().
2. The backing collection is now concurrent:
   private final Map<TopicPartition, PartitionState<K, V>> partitionStates = new ConcurrentHashMap<>();
   Its iterators are weakly consistent and do not throw ConcurrentModificationException.
3. getAssignedPartitions() collects into a fresh map before anything streams it:
   Collections.unmodifiableMap(partitionStates.entrySet().stream().filter(...).collect(Collectors.toMap(...))),
   so the traversal in getNumberOfIncompleteOffsets() runs over a snapshot, not the live map.

The HashMap$ValueSpliterator frame in the trace is therefore unreachable in current code.

Thanks to @JorgenRingen for the report - it was accurate, and it went unanswered for a year before
being closed without one.

_{Mirrored because the upstream issue was closed by an administrative sweep on 2023-07-07, not by a fix or a decision: all 28 issues closed that day carried the single comment "Closing Issue" and were marked completed. A separate sweep three weeks earlier (2023-06-15) closed 35 unmerged PRs with "Closing - Stale.", several of which implemented these issues. Mapping lives in src/docs/development/upstream-map.yaml.}