| commit | 3f07189a2a01dbc9c39282c1889d17e24f7f3f06 | [log] [tgz] |
|---|---|---|
| author | Etienne Chauchot <echauchot@apache.org> | Wed Jan 08 11:36:36 2020 +0100 |
| committer | Etienne Chauchot <echauchot@apache.org> | Thu Jan 23 16:10:06 2020 +0100 |
| tree | fb4fd3ca55ebf7cd97db24bfa99cf75d0d8227db | |
| parent | 367aefedf95bc03bae33ee05de2bfe8b1825be35 [diff] |
[BEAM-9065] Reset MetricsContainerStepMapAccumulator upon initialization of MetricsAccumulator singleton because metrics need to be reset when several pipelines are run inside the same JVM (cherry picked from commit fd3da1e68f18560954f9c736146c7565e4a1af0f)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/MetricsAccumulator.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/MetricsAccumulator.java index 2d83c3c..8313d73 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/MetricsAccumulator.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/MetricsAccumulator.java
@@ -74,6 +74,8 @@ } } LOG.info("Instantiated metrics accumulator: " + instance.value()); + } else { + instance.reset(); } }
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/MetricsAccumulator.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/MetricsAccumulator.java index f674ef9..e182cf0 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/MetricsAccumulator.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/MetricsAccumulator.java
@@ -51,6 +51,8 @@ } } LOG.info("Instantiated metrics accumulator: " + instance.value()); + } else { + instance.reset(); } }