You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by da...@apache.org on 2023/12/18 15:34:06 UTC
(beam) branch users/damccorm/releaseCP created (now 63bff87ff76)
This is an automated email from the ASF dual-hosted git repository.
damccorm pushed a change to branch users/damccorm/releaseCP
in repository https://gitbox.apache.org/repos/asf/beam.git
at 63bff87ff76 getProcessingTimesByStep returns a modifiable map (#29786)
This branch includes the following new commits:
new 63bff87ff76 getProcessingTimesByStep returns a modifiable map (#29786)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(beam) 01/01: getProcessingTimesByStep returns a modifiable map (#29786)
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
damccorm pushed a commit to branch users/damccorm/releaseCP
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 63bff87ff76ea3147e148d074f5983f5f181f0bd
Author: clmccart <cl...@gmail.com>
AuthorDate: Mon Dec 18 07:32:36 2023 -0800
getProcessingTimesByStep returns a modifiable map (#29786)
* getProcessingTimesByStep returns a modifiable map
* return copy in getProcessingTimesByStep and update name accordingly
* Spotless
---------
Co-authored-by: Claire McCarthy <cl...@google.com>
Co-authored-by: Danny McCormick <da...@google.com>
---
.../beam/runners/dataflow/worker/DataflowExecutionContext.java | 5 +++--
.../runners/dataflow/worker/DataflowExecutionStateSampler.java | 5 +++--
.../runners/dataflow/worker/DataflowExecutionContextTest.java | 2 +-
.../dataflow/worker/DataflowExecutionStateSamplerTest.java | 10 +++++-----
4 files changed, 12 insertions(+), 10 deletions(-)
diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContext.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContext.java
index b861295d8b9..7d45295b2d8 100644
--- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContext.java
+++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContext.java
@@ -341,8 +341,9 @@ public abstract class DataflowExecutionContext<T extends DataflowStepContext> {
return Optional.ofNullable(activeMessageMetadata);
}
- public Map<String, IntSummaryStatistics> getProcessingTimesByStep() {
- return Collections.unmodifiableMap(processingTimesByStep);
+ public Map<String, IntSummaryStatistics> getProcessingTimesByStepCopy() {
+ Map<String, IntSummaryStatistics> processingTimesCopy = processingTimesByStep;
+ return processingTimesCopy;
}
/**
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 c5c3b2d4180..61b7e01e49a 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
@@ -86,7 +86,8 @@ public final class DataflowExecutionStateSampler extends ExecutionStateSampler {
return;
}
DataflowExecutionStateTracker dfTracker = (DataflowExecutionStateTracker) tracker;
- completedProcessingMetrics.put(dfTracker.getWorkItemId(), dfTracker.getProcessingTimesByStep());
+ completedProcessingMetrics.put(
+ dfTracker.getWorkItemId(), dfTracker.getProcessingTimesByStepCopy());
activeTrackersByWorkId.remove(dfTracker.getWorkItemId());
// Attribute any remaining time since the last sampling while removing the tracker.
@@ -126,7 +127,7 @@ public final class DataflowExecutionStateSampler extends ExecutionStateSampler {
DataflowExecutionStateTracker tracker = activeTrackersByWorkId.get(workId);
return mergeStepStatsMaps(
completedProcessingMetrics.getOrDefault(workId, new HashMap<>()),
- tracker.getProcessingTimesByStep());
+ tracker.getProcessingTimesByStepCopy());
}
public void resetForWorkId(String workId) {
diff --git a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContextTest.java b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContextTest.java
index 3dc026ca7c8..01951c2f83e 100644
--- a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContextTest.java
+++ b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContextTest.java
@@ -176,7 +176,7 @@ public class DataflowExecutionContextTest {
tracker.enterState(newState);
// The first completed state should be recorded and the new state should be active.
- Map<String, IntSummaryStatistics> gotProcessingTimes = tracker.getProcessingTimesByStep();
+ Map<String, IntSummaryStatistics> gotProcessingTimes = tracker.getProcessingTimesByStepCopy();
Assert.assertEquals(1, gotProcessingTimes.size());
Assert.assertEquals(
new HashSet<>(Arrays.asList(NameContextsForTests.nameContextForTest().userName())),
diff --git a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSamplerTest.java b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSamplerTest.java
index 1377148bc20..b772952b74c 100644
--- a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSamplerTest.java
+++ b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionStateSamplerTest.java
@@ -76,7 +76,7 @@ public class DataflowExecutionStateSamplerTest {
Map<String, IntSummaryStatistics> testCompletedProcessingTimes = new HashMap<>();
testCompletedProcessingTimes.put("some-step", new IntSummaryStatistics());
DataflowExecutionStateTracker trackerMock = createMockTracker(workId);
- when(trackerMock.getProcessingTimesByStep()).thenReturn(testCompletedProcessingTimes);
+ when(trackerMock.getProcessingTimesByStepCopy()).thenReturn(testCompletedProcessingTimes);
sampler.addTracker(trackerMock);
sampler.removeTracker(trackerMock);
@@ -98,7 +98,7 @@ public class DataflowExecutionStateSamplerTest {
ActiveMessageMetadata.create(step1act1.getStepName().userName(), clock.getMillis());
DataflowExecutionStateTracker trackerMock = createMockTracker(workId);
when(trackerMock.getActiveMessageMetadata()).thenReturn(Optional.of(testMetadata));
- when(trackerMock.getProcessingTimesByStep()).thenReturn(testCompletedProcessingTimes);
+ when(trackerMock.getProcessingTimesByStepCopy()).thenReturn(testCompletedProcessingTimes);
sampler.addTracker(trackerMock);
@@ -122,13 +122,13 @@ public class DataflowExecutionStateSamplerTest {
equalTo(tracker1Mock.getActiveMessageMetadata()));
assertThat(
sampler.getProcessingDistributionsForWorkId(workId1),
- equalTo(tracker1Mock.getProcessingTimesByStep()));
+ equalTo(tracker1Mock.getProcessingTimesByStepCopy()));
assertThat(
sampler.getActiveMessageMetadataForWorkId(workId2),
equalTo(tracker2Mock.getActiveMessageMetadata()));
assertThat(
sampler.getProcessingDistributionsForWorkId(workId2),
- equalTo(tracker2Mock.getProcessingTimesByStep()));
+ equalTo(tracker2Mock.getProcessingTimesByStepCopy()));
sampler.removeTracker(tracker1Mock);
sampler.removeTracker(tracker2Mock);
@@ -136,7 +136,7 @@ public class DataflowExecutionStateSamplerTest {
assertThat(
sampler.getProcessingDistributionsForWorkId(workId1),
- equalTo(tracker1Mock.getProcessingTimesByStep()));
+ equalTo(tracker1Mock.getProcessingTimesByStepCopy()));
Assert.assertTrue(sampler.getProcessingDistributionsForWorkId(workId2).isEmpty());
}