clmccart commented on code in PR #30270:
URL: https://github.com/apache/beam/pull/30270#discussion_r1484568792
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowExecutionContext.java:
##########
@@ -341,8 +342,8 @@ public Optional<ActiveMessageMetadata>
getActiveMessageMetadata() {
return Optional.ofNullable(activeMessageMetadata);
}
- public Map<String, IntSummaryStatistics> getProcessingTimesByStepCopy() {
- Map<String, IntSummaryStatistics> processingTimesCopy =
processingTimesByStep;
+ public synchronized Map<String, IntSummaryStatistics>
getProcessingTimesByStepCopy() {
+ Map<String, IntSummaryStatistics> processingTimesCopy = new
HashMap<>(processingTimesByStep);
Review Comment:
the sampler will also access activeMessageMetadata so i think i do need to
synchronize that. since getActiveMessageMetadata returns an optional, i dont
need to copy it , correct?
as for the deep copy of the IntSummaryStatistics, it didn't like the clone.
compiler was throwing `'clone()' has protected access in 'java.lang.Object'`.
Would this work?
Map<String, IntSummaryStatistics> processingTimesCopy =
processingTimesByStep.entrySet().stream()
.collect(Collectors.toMap(e -> e.getKey(), e -> {
IntSummaryStatistics clone = new IntSummaryStatistics();
clone.combine(e.getValue());
return clone;
}));
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
