[BEAM-5759] ConcurrentModificationException on JmsIO checkpoint finalization

• Type: Bug
• Status: Resolved
• Resolution: Fixed
• Affects Version/s: 2.8.0
• Fix Version/s: 2.9.0
• Component/s: io-java-jms
• Assignee: Andrew Fulton
• Reporter: Andrew Fulton
• Created: 16/Oct/18 10:54
• Updated: 17/Oct/18 13:51
• Resolved: 17/Oct/18 13:51

Description

When reading from a JmsIO source, a ConcurrentModificationException can be thrown when checkpoint finalization occurs under heavy load.

For example:

jsonPayload: {
  exception: "java.util.ConcurrentModificationException
    at java.util.ArrayList$Itr.checkForComodification(ArrayList.java:903)
    at java.util.ArrayList$Itr.next(ArrayList.java:853)
    at org.apache.beam.sdk.io.jms.JmsCheckpointMark.finalizeCheckpoint(JmsCheckpointMark.java:65)
    at com.google.cloud.dataflow.worker.StreamingModeExecutionContext$1.run(StreamingModeExecutionContext.java:379)
    at com.google.cloud.dataflow.worker.StreamingDataflowWorker$8.run(StreamingDataflowWorker.java:846)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
"
  job: "2018-09-27_08_55_18-6454085774348718625"   
  logger: "com.google.cloud.dataflow.worker.StreamingDataflowWorker"   
  message: "Source checkpoint finalization failed:"   
  thread: "309"   
  work: "<nil>"   
  worker: "test-andrew-092715504-09270855-tkfp-harness-dnmb"   

 

Looking at the JmsCheckpointMark code, it appears that access to the pending message list is unprotected - thus if a thread calls finalizeCheckpoint while a separate processing thread adds more messages to the checkpoint mark list then an exception will be thrown.

Issue Links

• links to: GitHub Pull Request #6702