[Cherry Pick #29979] Fix race condition in Dataflow sampler. (#30144)

* make getProcessingDistributionsForWorkId synchronized to avoid race conditions with remove tracker

* correctly access the concurrent hashmap to avoid race conditions

---------

Co-authored-by: Claire McCarthy <clairemccarthy@google.com>

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java

@@ -118,13 +118,10 @@ public Optional<ActiveMessageMetadata> getActiveMessageMetadataForWorkId(String
   }
 
   public Map<String, IntSummaryStatistics> getProcessingDistributionsForWorkId(String workId) {
-    if (!activeTrackersByWorkId.containsKey(workId)) {
-      if (completedProcessingMetrics.containsKey(workId)) {
-        return completedProcessingMetrics.get(workId);
-      }
-      return new HashMap<>();
-    }
     DataflowExecutionStateTracker tracker = activeTrackersByWorkId.get(workId);
+    if (tracker == null) {
+      return completedProcessingMetrics.getOrDefault(workId, new HashMap<>());
+    }
     return mergeStepStatsMaps(
         completedProcessingMetrics.getOrDefault(workId, new HashMap<>()),
         tracker.getProcessingTimesByStepCopy());