You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lo...@apache.org on 2024/01/29 19:13:13 UTC
(beam) branch master updated: make getProcessingDistributionsForWorkId threadsafe (#29979)
This is an automated email from the ASF dual-hosted git repository.
lostluck pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new b72eacfba37 make getProcessingDistributionsForWorkId threadsafe (#29979)
b72eacfba37 is described below
commit b72eacfba37d8e5a2bc821b86b0bcf42900df087
Author: clmccart <cl...@gmail.com>
AuthorDate: Mon Jan 29 11:13:05 2024 -0800
make getProcessingDistributionsForWorkId threadsafe (#29979)
* make getProcessingDistributionsForWorkId synchronized to avoid race conditions with remove tracker
* correctly access the concurrent hashmap to avoid race conditions
---------
Co-authored-by: Claire McCarthy <cl...@google.com>
---
.../runners/dataflow/worker/DataflowExecutionStateSampler.java | 9 +++------
1 file changed, 3 insertions(+), 6 deletions(-)
diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java
index 61b7e01e49a..1ff9a9be40d 100644
--- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java
+++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSampler.java
@@ -118,13 +118,10 @@ public final class DataflowExecutionStateSampler extends ExecutionStateSampler {
}
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());