What happened?

There is a potential race condition in the DataflowExecutionStateSampler that occurs in getProcessingDistributionsForWorkId when multiple threads are modifying the activeTrackersByWorkId map. In getProcessingDistributionsForWorkId, we check for the existence of the work id in the activeTrackersByWorkId map before retrieving the value from the map. If another thread removes the tracker from the map in-between us checking for its existence and actually grabbing the object, we will get a null tracker that then throws a NullPointerException when we try to access data on it.

A stack trace like so will be dumped:

Error message from worker: java.lang.NullPointerException
        org.apache.beam.runners.dataflow.worker.DataflowExecutionStateSampler.getProcessingDistributionsForWorkId(DataflowExecutionStateSampler.java:130)
        org.apache.beam.runners.dataflow.worker.streaming.Work.addActiveLatencyBreakdownToBuilder(Work.java:166)
        org.apache.beam.runners.dataflow.worker.streaming.Work.getLatencyAttributions(Work.java:135)
        org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.process(StreamingDataflowWorker.java:1167)
        org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.lambda$scheduleWorkItem$5(StreamingDataflowWorker.java:873)
        org.apache.beam.runners.dataflow.worker.streaming.Work.run(Work.java:74)
        org.apache.beam.runners.dataflow.worker.util.BoundedQueueExecutor.lambda$executeLockHeld$0(BoundedQueueExecutor.java:183)
        java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
        java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
        java.base/java.lang.Thread.run(Thread.java:829)

The getProcessingDistributionsForWorkId method needs to be synchronized.

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner