[
https://issues.apache.org/jira/browse/BEAM-3892?focusedWorklogId=84366&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-84366
]
ASF GitHub Bot logged work on BEAM-3892:
----------------------------------------
Author: ASF GitHub Bot
Created on: 26/Mar/18 14:22
Start Date: 26/Mar/18 14:22
Worklog Time Spent: 10m
Work Description: echauchot closed pull request #4918: [BEAM-3892] Make
MetricQueryResults and related classes more json-serialization friendly
URL: https://github.com/apache/beam/pull/4918
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/metrics/MetricFiltering.java
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/metrics/MetricFiltering.java
index 99d8f0f38b7..b133e7d6601 100644
---
a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/metrics/MetricFiltering.java
+++
b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/metrics/MetricFiltering.java
@@ -91,8 +91,8 @@ private static boolean matchesName(MetricName metricName,
Set<MetricNameFilter>
return true;
}
for (MetricNameFilter nameFilter : nameFilters) {
- if ((nameFilter.getName() == null ||
nameFilter.getName().equals(metricName.name()))
- && Objects.equal(metricName.namespace(), nameFilter.getNamespace()))
{
+ if ((nameFilter.getName() == null ||
nameFilter.getName().equals(metricName.getName()))
+ && Objects.equal(metricName.getNamespace(),
nameFilter.getNamespace())) {
return true;
}
}
diff --git
a/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsContainerStepMap.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsContainerStepMap.java
index f643139f9bd..295e7c86fe2 100644
---
a/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsContainerStepMap.java
+++
b/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsContainerStepMap.java
@@ -115,9 +115,9 @@ public static MetricResults asMetricResults(
* Returns {@link MetricResults} based on given {@link
MetricsContainerStepMap} of attempted
* metrics.
*
- * <p>This constructor is intended for runners which only support
`attempted` metrics.
- * Accessing {@link MetricResult#committed()} in the resulting {@link
MetricResults} will result
- * in an {@link UnsupportedOperationException}.</p>
+ * <p>This constructor is intended for runners which only support
`attempted` metrics. Accessing
+ * {@link MetricResult#getCommitted()} in the resulting {@link
MetricResults} will result in an
+ * {@link UnsupportedOperationException}.
*/
public static MetricResults asAttemptedOnlyMetricResults(
MetricsContainerStepMap attemptedMetricsContainers) {
@@ -230,7 +230,7 @@ private QueryResults(MetricsFilter filter) {
}
@Override
- public Iterable<MetricResult<Long>> counters() {
+ public Iterable<MetricResult<Long>> getCounters() {
return
FluentIterable
.from(counters.values())
@@ -240,7 +240,7 @@ private QueryResults(MetricsFilter filter) {
}
@Override
- public Iterable<MetricResult<DistributionResult>> distributions() {
+ public Iterable<MetricResult<DistributionResult>> getDistributions() {
return
FluentIterable
.from(distributions.values())
@@ -250,7 +250,7 @@ private QueryResults(MetricsFilter filter) {
}
@Override
- public Iterable<MetricResult<GaugeResult>> gauges() {
+ public Iterable<MetricResult<GaugeResult>> getGauges() {
return
FluentIterable
.from(gauges.values())
@@ -401,17 +401,17 @@ private AccumulatedMetricResult(
}
@Override
- public MetricName name() {
+ public MetricName getName() {
return name;
}
@Override
- public String step() {
+ public String getStep() {
return step;
}
@Override
- public T committed() {
+ public T getCommitted() {
if (!isCommittedSupported) {
throw new UnsupportedOperationException("This runner does not
currently support committed"
+ " metrics results. Please use 'attempted' instead.");
@@ -420,7 +420,7 @@ public T committed() {
}
@Override
- public T attempted() {
+ public T getAttempted() {
return attempted;
}
}
diff --git
a/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsTranslation.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsTranslation.java
index 596fe231fcd..5c000ab684e 100644
---
a/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsTranslation.java
+++
b/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MetricsTranslation.java
@@ -137,8 +137,8 @@ public static MetricName
metricNameFromProto(BeamFnApi.Metrics.User.MetricName p
public static BeamFnApi.Metrics.User.MetricName metricNameToProto(MetricName
metricName) {
return BeamFnApi.Metrics.User.MetricName.newBuilder()
- .setNamespace(metricName.namespace())
- .setName(metricName.name())
+ .setNamespace(metricName.getNamespace())
+ .setName(metricName.getName())
.build();
}
}
diff --git
a/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/MetricUpdateMatchers.java
b/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/MetricUpdateMatchers.java
index a5ee874feb9..1b8dc7ad5f8 100644
---
a/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/MetricUpdateMatchers.java
+++
b/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/MetricUpdateMatchers.java
@@ -38,7 +38,7 @@
return new TypeSafeMatcher<MetricUpdate<T>>() {
@Override
protected boolean matchesSafely(MetricUpdate<T> item) {
- return Objects.equals(name, item.getKey().metricName().name())
+ return Objects.equals(name, item.getKey().metricName().getName())
&& Objects.equals(update, item.getUpdate());
}
@@ -62,8 +62,8 @@ public void describeTo(Description description) {
return new TypeSafeMatcher<MetricUpdate<T>>() {
@Override
protected boolean matchesSafely(MetricUpdate<T> item) {
- return Objects.equals(namespace,
item.getKey().metricName().namespace())
- && Objects.equals(name, item.getKey().metricName().name())
+ return Objects.equals(namespace,
item.getKey().metricName().getNamespace())
+ && Objects.equals(name, item.getKey().metricName().getName())
&& Objects.equals(step, item.getKey().stepName())
&& Objects.equals(update, item.getUpdate());
}
diff --git
a/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/MetricsContainerStepMapTest.java
b/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/MetricsContainerStepMapTest.java
index f5f7cafedbe..c8b172b0e8d 100644
---
a/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/MetricsContainerStepMapTest.java
+++
b/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/MetricsContainerStepMapTest.java
@@ -93,9 +93,9 @@ public void testAttemptedAccumulatedMetricResults() {
MetricQueryResults step1res =
metricResults.queryMetrics(MetricsFilter.builder().addStep(STEP1).build());
- assertIterableSize(step1res.counters(), 1);
- assertIterableSize(step1res.distributions(), 1);
- assertIterableSize(step1res.gauges(), 1);
+ assertIterableSize(step1res.getCounters(), 1);
+ assertIterableSize(step1res.getDistributions(), 1);
+ assertIterableSize(step1res.getGauges(), 1);
assertCounter(COUNTER_NAME, step1res, STEP1, VALUE, false);
assertDistribution(DISTRIBUTION_NAME,
@@ -106,9 +106,9 @@ public void testAttemptedAccumulatedMetricResults() {
MetricQueryResults step2res =
metricResults.queryMetrics(MetricsFilter.builder().addStep(STEP2).build());
- assertIterableSize(step2res.counters(), 1);
- assertIterableSize(step2res.distributions(), 1);
- assertIterableSize(step2res.gauges(), 1);
+ assertIterableSize(step2res.getCounters(), 1);
+ assertIterableSize(step2res.getDistributions(), 1);
+ assertIterableSize(step2res.getGauges(), 1);
assertCounter(COUNTER_NAME, step2res, STEP2, VALUE * 2, false);
assertDistribution(
@@ -120,9 +120,9 @@ public void testAttemptedAccumulatedMetricResults() {
MetricQueryResults allres =
metricResults.queryMetrics(MetricsFilter.builder().build());
- assertIterableSize(allres.counters(), 2);
- assertIterableSize(allres.distributions(), 2);
- assertIterableSize(allres.gauges(), 2);
+ assertIterableSize(allres.getCounters(), 2);
+ assertIterableSize(allres.getDistributions(), 2);
+ assertIterableSize(allres.getGauges(), 2);
}
@Test
@@ -194,9 +194,9 @@ public void
testAttemptedAndCommittedAccumulatedMetricResults() {
MetricQueryResults step1res =
metricResults.queryMetrics(MetricsFilter.builder().addStep(STEP1).build());
- assertIterableSize(step1res.counters(), 1);
- assertIterableSize(step1res.distributions(), 1);
- assertIterableSize(step1res.gauges(), 1);
+ assertIterableSize(step1res.getCounters(), 1);
+ assertIterableSize(step1res.getDistributions(), 1);
+ assertIterableSize(step1res.getGauges(), 1);
assertCounter(COUNTER_NAME, step1res, STEP1, VALUE * 2, false);
assertDistribution(
@@ -212,9 +212,9 @@ public void
testAttemptedAndCommittedAccumulatedMetricResults() {
MetricQueryResults step2res =
metricResults.queryMetrics(MetricsFilter.builder().addStep(STEP2).build());
- assertIterableSize(step2res.counters(), 1);
- assertIterableSize(step2res.distributions(), 1);
- assertIterableSize(step2res.gauges(), 1);
+ assertIterableSize(step2res.getCounters(), 1);
+ assertIterableSize(step2res.getDistributions(), 1);
+ assertIterableSize(step2res.getGauges(), 1);
assertCounter(COUNTER_NAME, step2res, STEP2, VALUE * 3, false);
assertDistribution(DISTRIBUTION_NAME, step2res, STEP2,
@@ -229,9 +229,9 @@ public void
testAttemptedAndCommittedAccumulatedMetricResults() {
MetricQueryResults allres =
metricResults.queryMetrics(MetricsFilter.builder().build());
- assertIterableSize(allres.counters(), 2);
- assertIterableSize(allres.distributions(), 2);
- assertIterableSize(allres.gauges(), 2);
+ assertIterableSize(allres.getCounters(), 2);
+ assertIterableSize(allres.getDistributions(), 2);
+ assertIterableSize(allres.getGauges(), 2);
}
private <T> void assertIterableSize(Iterable<T> iterable, int size) {
@@ -245,7 +245,7 @@ private void assertCounter(
Long expected,
boolean isCommitted) {
assertThat(
- metricQueryResults.counters(),
+ metricQueryResults.getCounters(),
hasItem(metricsResult(NAMESPACE, name, step, expected, isCommitted)));
}
@@ -256,7 +256,7 @@ private void assertDistribution(
DistributionResult expected,
boolean isCommitted) {
assertThat(
- metricQueryResults.distributions(),
+ metricQueryResults.getDistributions(),
hasItem(metricsResult(NAMESPACE, name, step, expected, isCommitted)));
}
@@ -267,7 +267,7 @@ private void assertGauge(
GaugeResult expected,
boolean isCommitted) {
assertThat(
- metricQueryResults.gauges(),
+ metricQueryResults.getGauges(),
hasItem(metricsResult(NAMESPACE, name, step, expected, isCommitted)));
}
}
diff --git
a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectMetrics.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectMetrics.java
index 3ffbba35949..4ef1c61ffbf 100644
---
a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectMetrics.java
+++
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectMetrics.java
@@ -245,10 +245,10 @@ public static MetricQueryResults create(
abstract static class DirectMetricResult<T> implements MetricResult<T> {
// need to define these here so they appear in the correct order
// and the generated constructor is usable and consistent
- public abstract MetricName name();
- public abstract String step();
- public abstract T committed();
- public abstract T attempted();
+ public abstract MetricName getName();
+ public abstract String getStep();
+ public abstract T getCommitted();
+ public abstract T getAttempted();
public static <T> MetricResult<T> create(MetricName name, String scope,
T committed, T attempted) {
diff --git
a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectMetricsTest.java
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectMetricsTest.java
index f34bb0c34b6..751fc942bed 100644
---
a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectMetricsTest.java
+++
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectMetricsTest.java
@@ -92,22 +92,22 @@ public void testApplyCommittedNoFilter() {
));
MetricQueryResults results =
metrics.queryMetrics(MetricsFilter.builder().build());
- assertThat(results.counters(), containsInAnyOrder(
+ assertThat(results.getCounters(), containsInAnyOrder(
attemptedMetricsResult("ns1", "name1", "step1", 0L),
attemptedMetricsResult("ns1", "name2", "step1", 0L),
attemptedMetricsResult("ns1", "name1", "step2", 0L)));
- assertThat(results.counters(), containsInAnyOrder(
+ assertThat(results.getCounters(), containsInAnyOrder(
committedMetricsResult("ns1", "name1", "step1", 5L),
committedMetricsResult("ns1", "name2", "step1", 12L),
committedMetricsResult("ns1", "name1", "step2", 7L)));
- assertThat(results.distributions(), contains(
+ assertThat(results.getDistributions(), contains(
attemptedMetricsResult("ns1", "name1", "step1",
DistributionResult.IDENTITY_ELEMENT)));
- assertThat(results.distributions(), contains(
+ assertThat(results.getDistributions(), contains(
committedMetricsResult("ns1", "name1", "step1",
DistributionResult.create(12, 3, 3, 5))));
- assertThat(results.gauges(), contains(
+ assertThat(results.getGauges(), contains(
attemptedMetricsResult("ns2", "name2", "step1", GaugeResult.empty())
));
- assertThat(results.gauges(), contains(
+ assertThat(results.getGauges(), contains(
committedMetricsResult("ns2", "name2", "step1",
GaugeResult.create(27L, Instant.now()))
));
}
@@ -135,12 +135,12 @@ public void testApplyAttemptedCountersQueryOneNamespace()
{
MetricQueryResults results = metrics.queryMetrics(
MetricsFilter.builder().addNameFilter(inNamespace("ns1")).build());
- assertThat(results.counters(),
+ assertThat(results.getCounters(),
containsInAnyOrder(
attemptedMetricsResult("ns1", "name1", "step1", 5L),
attemptedMetricsResult("ns1", "name1", "step2", 7L)));
- assertThat(results.counters(),
+ assertThat(results.getCounters(),
containsInAnyOrder(
committedMetricsResult("ns1", "name1", "step1", 0L),
committedMetricsResult("ns1", "name1", "step2", 0L)));
@@ -169,12 +169,12 @@ public void testApplyAttemptedQueryCompositeScope() {
MetricQueryResults results = metrics.queryMetrics(
MetricsFilter.builder().addStep("Outer1").build());
- assertThat(results.counters(),
+ assertThat(results.getCounters(),
containsInAnyOrder(
attemptedMetricsResult("ns1", "name1", "Outer1/Inner1", 12L),
attemptedMetricsResult("ns1", "name1", "Outer1/Inner2", 8L)));
- assertThat(results.counters(),
+ assertThat(results.getCounters(),
containsInAnyOrder(
committedMetricsResult("ns1", "name1", "Outer1/Inner1", 0L),
committedMetricsResult("ns1", "name1", "Outer1/Inner2", 0L)));
@@ -204,7 +204,7 @@ public void testPartialScopeMatchingInMetricsQuery() {
MetricQueryResults results = metrics.queryMetrics(
MetricsFilter.builder().addStep("Top1/Outer1").build());
- assertThat(results.counters(),
+ assertThat(results.getCounters(),
containsInAnyOrder(
attemptedMetricsResult("ns1", "name1", "Top1/Outer1/Inner1", 5L),
attemptedMetricsResult("ns1", "name1", "Top1/Outer1/Inner2", 8L)));
@@ -212,7 +212,7 @@ public void testPartialScopeMatchingInMetricsQuery() {
results = metrics.queryMetrics(
MetricsFilter.builder().addStep("Inner2").build());
- assertThat(results.counters(),
+ assertThat(results.getCounters(),
containsInAnyOrder(
attemptedMetricsResult("ns1", "name1", "Top1/Outer1/Inner2", 8L),
attemptedMetricsResult("ns1", "name1", "Top1/Outer2/Inner2",
18L)));
diff --git
a/runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/FlinkMetricContainer.java
b/runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/FlinkMetricContainer.java
index 4defc382db6..cc61f3257c3 100644
---
a/runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/FlinkMetricContainer.java
+++
b/runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/FlinkMetricContainer.java
@@ -88,16 +88,16 @@ void updateMetrics() {
asAttemptedOnlyMetricResults(metricsAccumulator.getLocalValue());
MetricQueryResults metricQueryResults =
metricResults.queryMetrics(MetricsFilter.builder().build());
- updateCounters(metricQueryResults.counters());
- updateDistributions(metricQueryResults.distributions());
- updateGauge(metricQueryResults.gauges());
+ updateCounters(metricQueryResults.getCounters());
+ updateDistributions(metricQueryResults.getDistributions());
+ updateGauge(metricQueryResults.getGauges());
}
private void updateCounters(Iterable<MetricResult<Long>> counters) {
for (MetricResult<Long> metricResult : counters) {
String flinkMetricName = getFlinkMetricNameString(COUNTER_PREFIX,
metricResult);
- Long update = metricResult.attempted();
+ Long update = metricResult.getAttempted();
// update flink metric
Counter counter =
@@ -113,7 +113,7 @@ private void
updateDistributions(Iterable<MetricResult<DistributionResult>> dist
String flinkMetricName =
getFlinkMetricNameString(DISTRIBUTION_PREFIX, metricResult);
- DistributionResult update = metricResult.attempted();
+ DistributionResult update = metricResult.getAttempted();
// update flink metric
FlinkDistributionGauge gauge =
flinkDistributionGaugeCache.get(flinkMetricName);
@@ -132,7 +132,7 @@ private void
updateGauge(Iterable<MetricResult<GaugeResult>> gauges) {
String flinkMetricName =
getFlinkMetricNameString(GAUGE_PREFIX, metricResult);
- GaugeResult update = metricResult.attempted();
+ GaugeResult update = metricResult.getAttempted();
// update flink metric
FlinkGauge gauge = flinkGaugeCache.get(flinkMetricName);
@@ -148,9 +148,9 @@ private void
updateGauge(Iterable<MetricResult<GaugeResult>> gauges) {
private static String getFlinkMetricNameString(String prefix,
MetricResult<?> metricResult) {
return prefix
- + METRIC_KEY_SEPARATOR + metricResult.step()
- + METRIC_KEY_SEPARATOR + metricResult.name().namespace()
- + METRIC_KEY_SEPARATOR + metricResult.name().name();
+ + METRIC_KEY_SEPARATOR + metricResult.getStep()
+ + METRIC_KEY_SEPARATOR + metricResult.getName().getNamespace()
+ + METRIC_KEY_SEPARATOR + metricResult.getName().getName();
}
/**
diff --git
a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowMetrics.java
b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowMetrics.java
index e6b51bb1e0f..a7c83c2e920 100644
---
a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowMetrics.java
+++
b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowMetrics.java
@@ -311,7 +311,7 @@ public MetricQueryResults build() {
DataflowMetricResultExtractor extractor = new
DataflowMetricResultExtractor(
dataflowPipelineJob.getDataflowOptions().isStreaming());
for (MetricKey metricKey : metricHashKeys) {
- String metricName = metricKey.metricName().name();
+ String metricName = metricKey.metricName().getName();
if (metricName.endsWith("[MIN]") || metricName.endsWith("[MAX]")
|| metricName.endsWith("[MEAN]") ||
metricName.endsWith("[COUNT]")) {
// Skip distribution metrics, as these are not yet properly
supported.
@@ -345,13 +345,13 @@ public static MetricQueryResults create(
abstract static class DataflowMetricResult<T> implements MetricResult<T> {
// need to define these here so they appear in the correct order
// and the generated constructor is usable and consistent
- public abstract MetricName name();
- public abstract String step();
+ public abstract MetricName getName();
+ public abstract String getStep();
@Nullable
protected abstract T committedInternal();
- public abstract T attempted();
+ public abstract T getAttempted();
- public T committed() {
+ public T getCommitted() {
T committed = committedInternal();
if (committed == null) {
throw new UnsupportedOperationException("This runner does not
currently support committed"
diff --git
a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowMetricsTest.java
b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowMetricsTest.java
index 75ec6176f12..d8e8e5ed4af 100644
---
a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowMetricsTest.java
+++
b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowMetricsTest.java
@@ -112,8 +112,8 @@ public void testEmptyMetricUpdates() throws IOException {
DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient);
MetricQueryResults result = dataflowMetrics.queryMetrics(null);
- assertThat(ImmutableList.copyOf(result.counters()), is(empty()));
- assertThat(ImmutableList.copyOf(result.distributions()), is(empty()));
+ assertThat(ImmutableList.copyOf(result.getCounters()), is(empty()));
+ assertThat(ImmutableList.copyOf(result.getDistributions()), is(empty()));
}
@Test
@@ -209,9 +209,9 @@ public void testSingleCounterUpdates() throws IOException {
DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient);
MetricQueryResults result = dataflowMetrics.queryMetrics(null);
- assertThat(result.counters(), containsInAnyOrder(
+ assertThat(result.getCounters(), containsInAnyOrder(
attemptedMetricsResult("counterNamespace", "counterName",
"myStepName", 1234L)));
- assertThat(result.counters(), containsInAnyOrder(
+ assertThat(result.getCounters(), containsInAnyOrder(
committedMetricsResult("counterNamespace", "counterName",
"myStepName", 1234L)));
}
@@ -242,9 +242,9 @@ public void testIgnoreDistributionButGetCounterUpdates()
throws IOException {
DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient);
MetricQueryResults result = dataflowMetrics.queryMetrics(null);
- assertThat(result.counters(), containsInAnyOrder(
+ assertThat(result.getCounters(), containsInAnyOrder(
attemptedMetricsResult("counterNamespace", "counterName",
"myStepName", 1233L)));
- assertThat(result.counters(), containsInAnyOrder(
+ assertThat(result.getCounters(), containsInAnyOrder(
committedMetricsResult("counterNamespace", "counterName",
"myStepName", 1233L)));
}
@@ -275,10 +275,10 @@ public void testDistributionUpdates() throws IOException {
DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient);
MetricQueryResults result = dataflowMetrics.queryMetrics(null);
- assertThat(result.distributions(), contains(
+ assertThat(result.getDistributions(), contains(
attemptedMetricsResult("distributionNamespace", "distributionName",
"myStepName",
DistributionResult.create(18, 2, 2, 16))));
- assertThat(result.distributions(), contains(
+ assertThat(result.getDistributions(), contains(
committedMetricsResult("distributionNamespace", "distributionName",
"myStepName",
DistributionResult.create(18, 2, 2, 16))));
}
@@ -311,14 +311,14 @@ public void testDistributionUpdatesStreaming() throws
IOException {
DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient);
MetricQueryResults result = dataflowMetrics.queryMetrics(null);
try {
- result.distributions().iterator().next().committed();
+ result.getDistributions().iterator().next().getCommitted();
fail("Expected UnsupportedOperationException");
} catch (UnsupportedOperationException expected) {
assertThat(expected.getMessage(),
containsString("This runner does not currently support committed"
+ " metrics results. Please use 'attempted' instead."));
}
- assertThat(result.distributions(), contains(
+ assertThat(result.getDistributions(), contains(
attemptedMetricsResult("distributionNamespace", "distributionName",
"myStepName",
DistributionResult.create(18, 2, 2, 16))));
}
@@ -362,11 +362,11 @@ public void testMultipleCounterUpdates() throws
IOException {
DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient);
MetricQueryResults result = dataflowMetrics.queryMetrics(null);
- assertThat(result.counters(), containsInAnyOrder(
+ assertThat(result.getCounters(), containsInAnyOrder(
attemptedMetricsResult("counterNamespace", "counterName",
"myStepName", 1233L),
attemptedMetricsResult("otherNamespace", "otherCounter",
"myStepName3", 12L),
attemptedMetricsResult("otherNamespace", "counterName", "myStepName4",
1200L)));
- assertThat(result.counters(), containsInAnyOrder(
+ assertThat(result.getCounters(), containsInAnyOrder(
committedMetricsResult("counterNamespace", "counterName",
"myStepName", 1233L),
committedMetricsResult("otherNamespace", "otherCounter",
"myStepName3", 12L),
committedMetricsResult("otherNamespace", "counterName", "myStepName4",
1200L)));
@@ -408,14 +408,14 @@ public void testMultipleCounterUpdatesStreaming() throws
IOException {
DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient);
MetricQueryResults result = dataflowMetrics.queryMetrics(null);
try {
- result.counters().iterator().next().committed();
+ result.getCounters().iterator().next().getCommitted();
fail("Expected UnsupportedOperationException");
} catch (UnsupportedOperationException expected) {
assertThat(expected.getMessage(),
containsString("This runner does not currently support committed"
+ " metrics results. Please use 'attempted' instead."));
}
- assertThat(result.counters(), containsInAnyOrder(
+ assertThat(result.getCounters(), containsInAnyOrder(
attemptedMetricsResult("counterNamespace", "counterName",
"myStepName", 1233L),
attemptedMetricsResult("otherNamespace", "otherCounter",
"myStepName3", 12L),
attemptedMetricsResult("otherNamespace", "counterName", "myStepName4",
1200L)));
diff --git
a/runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/SparkBeamMetric.java
b/runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/SparkBeamMetric.java
index 5e67445931a..4b0484a8169 100644
---
a/runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/SparkBeamMetric.java
+++
b/runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/SparkBeamMetric.java
@@ -48,31 +48,31 @@
MetricsAccumulator.getInstance().value());
MetricQueryResults metricQueryResults =
metricResults.queryMetrics(MetricsFilter.builder().build());
- for (MetricResult<Long> metricResult : metricQueryResults.counters()) {
- metrics.put(renderName(metricResult), metricResult.attempted());
+ for (MetricResult<Long> metricResult : metricQueryResults.getCounters()) {
+ metrics.put(renderName(metricResult), metricResult.getAttempted());
}
- for (MetricResult<DistributionResult> metricResult :
metricQueryResults.distributions()) {
- DistributionResult result = metricResult.attempted();
- metrics.put(renderName(metricResult) + ".count", result.count());
- metrics.put(renderName(metricResult) + ".sum", result.sum());
- metrics.put(renderName(metricResult) + ".min", result.min());
- metrics.put(renderName(metricResult) + ".max", result.max());
- metrics.put(renderName(metricResult) + ".mean", result.mean());
+ for (MetricResult<DistributionResult> metricResult :
metricQueryResults.getDistributions()) {
+ DistributionResult result = metricResult.getAttempted();
+ metrics.put(renderName(metricResult) + ".count", result.getCount());
+ metrics.put(renderName(metricResult) + ".sum", result.getSum());
+ metrics.put(renderName(metricResult) + ".min", result.getMin());
+ metrics.put(renderName(metricResult) + ".max", result.getMax());
+ metrics.put(renderName(metricResult) + ".mean", result.getMean());
}
- for (MetricResult<GaugeResult> metricResult : metricQueryResults.gauges())
{
- metrics.put(renderName(metricResult), metricResult.attempted().value());
+ for (MetricResult<GaugeResult> metricResult :
metricQueryResults.getGauges()) {
+ metrics.put(renderName(metricResult),
metricResult.getAttempted().getValue());
}
return metrics;
}
@VisibleForTesting
String renderName(MetricResult<?> metricResult) {
- String renderedStepName =
metricResult.step().replaceAll(ILLEGAL_CHARACTERS_AND_PERIOD, "_");
+ String renderedStepName =
metricResult.getStep().replaceAll(ILLEGAL_CHARACTERS_AND_PERIOD, "_");
if (renderedStepName.endsWith("_")) {
renderedStepName = renderedStepName.substring(0,
renderedStepName.length() - 1);
}
- MetricName metricName = metricResult.name();
- return (renderedStepName + "." + metricName.namespace() + "." +
metricName.name())
+ MetricName metricName = metricResult.getName();
+ return (renderedStepName + "." + metricName.getNamespace() + "." +
metricName.getName())
.replaceAll(ILLEGAL_CHARACTERS, "_");
}
}
diff --git
a/runners/spark/src/test/java/org/apache/beam/runners/spark/metrics/SparkBeamMetricTest.java
b/runners/spark/src/test/java/org/apache/beam/runners/spark/metrics/SparkBeamMetricTest.java
index 9426b2c2cf9..b3745aaf7f2 100644
---
a/runners/spark/src/test/java/org/apache/beam/runners/spark/metrics/SparkBeamMetricTest.java
+++
b/runners/spark/src/test/java/org/apache/beam/runners/spark/metrics/SparkBeamMetricTest.java
@@ -34,22 +34,22 @@
public void testRenderName() throws Exception {
MetricResult<Object> metricResult = new MetricResult<Object>() {
@Override
- public MetricName name() {
+ public MetricName getName() {
return MetricName.named("myNameSpace//", "myName()");
}
@Override
- public String step() {
+ public String getStep() {
return "myStep.one.two(three)";
}
@Override
- public Object committed() {
+ public Object getCommitted() {
return null;
}
@Override
- public Object attempted() {
+ public Object getAttempted() {
return null;
}
};
diff --git
a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/ResumeFromCheckpointStreamingTest.java
b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/ResumeFromCheckpointStreamingTest.java
index c509f93620f..d8363a33de1 100644
---
a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/ResumeFromCheckpointStreamingTest.java
+++
b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/ResumeFromCheckpointStreamingTest.java
@@ -162,10 +162,10 @@ public void testWithResume() throws Exception {
// first run should expect EOT matching the last injected element.
SparkPipelineResult res = run(Optional.of(new Instant(400)), 0);
- assertThat(res.metrics().queryMetrics(metricsFilter).counters(),
+ assertThat(res.metrics().queryMetrics(metricsFilter).getCounters(),
hasItem(attemptedMetricsResult(ResumeFromCheckpointStreamingTest.class.getName(),
"allMessages", "EOFShallNotPassFn", 4L)));
- assertThat(res.metrics().queryMetrics(metricsFilter).counters(),
+ assertThat(res.metrics().queryMetrics(metricsFilter).getCounters(),
hasItem(attemptedMetricsResult(ResumeFromCheckpointStreamingTest.class.getName(),
"processedMessages", "EOFShallNotPassFn", 4L)));
@@ -183,10 +183,10 @@ public void testWithResume() throws Exception {
// recovery should resume from last read offset, and read the second batch
of input.
res = runAgain(1);
// assertions 2:
- assertThat(res.metrics().queryMetrics(metricsFilter).counters(),
+ assertThat(res.metrics().queryMetrics(metricsFilter).getCounters(),
hasItem(attemptedMetricsResult(ResumeFromCheckpointStreamingTest.class.getName(),
"processedMessages", "EOFShallNotPassFn", 5L)));
- assertThat(res.metrics().queryMetrics(metricsFilter).counters(),
+ assertThat(res.metrics().queryMetrics(metricsFilter).getCounters(),
hasItem(attemptedMetricsResult(ResumeFromCheckpointStreamingTest.class.getName(),
"allMessages", "EOFShallNotPassFn", 6L)));
long successAssertions = 0;
@@ -194,9 +194,9 @@ public void testWithResume() throws Exception {
MetricsFilter.builder()
.addNameFilter(
MetricNameFilter.named(PAssertWithoutFlatten.class,
PAssert.SUCCESS_COUNTER))
- .build()).counters();
+ .build()).getCounters();
for (MetricResult<Long> counter : counterResults) {
- if (counter.attempted() > 0) {
+ if (counter.getAttempted() > 0) {
successAssertions++;
}
}
@@ -211,9 +211,9 @@ public void testWithResume() throws Exception {
MetricsFilter.builder()
.addNameFilter(MetricNameFilter.named(
PAssertWithoutFlatten.class, PAssert.FAILURE_COUNTER))
- .build()).counters();
+ .build()).getCounters();
for (MetricResult<Long> counter : failCounterResults) {
- if (counter.attempted() > 0) {
+ if (counter.getAttempted() > 0) {
failedAssertions++;
}
}
diff --git
a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/StreamingSourceMetricsTest.java
b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/StreamingSourceMetricsTest.java
index cafd539f7fc..e3c23d232c8 100644
---
a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/StreamingSourceMetricsTest.java
+++
b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/StreamingSourceMetricsTest.java
@@ -65,13 +65,14 @@ public void testUnboundedSourceMetrics() {
.queryMetrics(
MetricsFilter.builder()
.addNameFilter(
- MetricNameFilter.named(ELEMENTS_READ.namespace(),
ELEMENTS_READ.name()))
+ MetricNameFilter.named(
+ ELEMENTS_READ.getNamespace(),
ELEMENTS_READ.getName()))
.build());
- assertThat(metrics.counters(), hasItem(
+ assertThat(metrics.getCounters(), hasItem(
attemptedMetricsResult(
- ELEMENTS_READ.namespace(),
- ELEMENTS_READ.name(),
+ ELEMENTS_READ.getNamespace(),
+ ELEMENTS_READ.getName(),
"Read(UnboundedCountingSource)",
1000L)));
}
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DistributionResult.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DistributionResult.java
index 6da721068fb..600a49453d4 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DistributionResult.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DistributionResult.java
@@ -28,13 +28,13 @@
@AutoValue
public abstract class DistributionResult {
- public abstract long sum();
- public abstract long count();
- public abstract long min();
- public abstract long max();
+ public abstract long getSum();
+ public abstract long getCount();
+ public abstract long getMin();
+ public abstract long getMax();
- public double mean() {
- return (1.0 * sum()) / count();
+ public double getMean() {
+ return (1.0 * getSum()) / getCount();
}
/** The IDENTITY_ELEMENT is used to start accumulating distributions. */
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/GaugeResult.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/GaugeResult.java
index f24ded213dc..7bcf8548748 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/GaugeResult.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/GaugeResult.java
@@ -28,9 +28,9 @@
@Experimental(Kind.METRICS)
@AutoValue
public abstract class GaugeResult {
- public abstract long value();
+ public abstract long getValue();
- public abstract Instant timestamp();
+ public abstract Instant getTimestamp();
public static GaugeResult create(long value, Instant timestamp) {
return new AutoValue_GaugeResult(value, timestamp);
@@ -52,12 +52,12 @@ private EmptyGaugeResult() {
}
@Override
- public long value() {
+ public long getValue() {
return -1L;
}
@Override
- public Instant timestamp() {
+ public Instant getTimestamp() {
return EPOCH;
}
}
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricName.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricName.java
index 6c88fa2b1cf..56d897a2863 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricName.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricName.java
@@ -26,19 +26,19 @@
import org.apache.beam.sdk.annotations.Experimental.Kind;
/**
- * The name of a metric consists of a {@link #namespace} and a {@link #name}.
The {@link #namespace}
- * allows grouping related metrics together and also prevents collisions
between multiple metrics
- * with the same name.
+ * The name of a metric consists of a {@link #getNamespace} and a {@link
#getName}. The {@link
+ * #getNamespace} allows grouping related metrics together and also prevents
collisions between
+ * multiple metrics with the same name.
*/
@Experimental(Kind.METRICS)
@AutoValue
public abstract class MetricName implements Serializable {
/** The namespace associated with this metric. */
- public abstract String namespace();
+ public abstract String getNamespace();
/** The name of this metric. */
- public abstract String name();
+ public abstract String getName();
public static MetricName named(String namespace, String name) {
checkArgument(!Strings.isNullOrEmpty(namespace), "Metric namespace must be
non-empty");
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricQueryResults.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricQueryResults.java
index a7838eef33a..f43a9362a20 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricQueryResults.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricQueryResults.java
@@ -26,11 +26,11 @@
@Experimental(Kind.METRICS)
public interface MetricQueryResults {
/** Return the metric results for the counters that matched the filter. */
- Iterable<MetricResult<Long>> counters();
+ Iterable<MetricResult<Long>> getCounters();
/** Return the metric results for the distributions that matched the filter.
*/
- Iterable<MetricResult<DistributionResult>> distributions();
+ Iterable<MetricResult<DistributionResult>> getDistributions();
/** Return the metric results for the gauges that matched the filter. */
- Iterable<MetricResult<GaugeResult>> gauges();
+ Iterable<MetricResult<GaugeResult>> getGauges();
}
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricResult.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricResult.java
index 9a3971a931f..3619a6a3078 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricResult.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricResult.java
@@ -26,9 +26,9 @@
@Experimental(Kind.METRICS)
public interface MetricResult<T> {
/** Return the name of the metric. */
- MetricName name();
+ MetricName getName();
/** Return the step context to which this metric result applies. */
- String step();
+ String getStep();
/**
* Return the value of this metric across all successfully completed parts
of the pipeline.
@@ -36,10 +36,10 @@
* <p>Not all runners will support committed metrics. If they are not
supported, the runner will
* throw an {@link UnsupportedOperationException}.
*/
- T committed();
+ T getCommitted();
/**
* Return the value of this metric across all attempts of executing all
parts of the pipeline.
*/
- T attempted();
+ T getAttempted();
}
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricResults.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricResults.java
index ce751d4e23a..a745a6bef6e 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricResults.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricResults.java
@@ -35,8 +35,8 @@
*
* <p>For each type of metric, the result contains an iterable of all
metrics of that type that
* matched the filter. Each {@link MetricResult} includes the name of the
metric, the step in
- * which it was reported and the {@link MetricResult#committed} and
- * {@link MetricResult#attempted} values.
+ * which it was reported and the {@link MetricResult#getCommitted} and
+ * {@link MetricResult#getAttempted} values.
*
* <p>Note that runners differ in their support for committed and attempted
values.
*
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipeline.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipeline.java
index a6b8e02f4d8..0df5fdbc0a0 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipeline.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipeline.java
@@ -494,9 +494,9 @@ public static void verifyPAssertsSucceeded(Pipeline
pipeline, PipelineResult pip
MetricsFilter.builder()
.addNameFilter(MetricNameFilter.named(PAssert.class,
PAssert.SUCCESS_COUNTER))
.build())
- .counters();
+ .getCounters();
for (MetricResult<Long> counter : successCounterResults) {
- if (counter.attempted() > 0) {
+ if (counter.getAttempted() > 0) {
successfulAssertions++;
}
}
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesAttemptedMetrics.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesAttemptedMetrics.java
index 0a17980b0a4..0b28d1e8cf2 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesAttemptedMetrics.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesAttemptedMetrics.java
@@ -23,6 +23,6 @@
/**
* Category tag for validation tests which utilize {@link
org.apache.beam.sdk.metrics.Metrics}.
* Tests tagged with {@link UsesAttemptedMetrics} should be run for runners
which support
- * {@link MetricResult#attempted()}.
+ * {@link MetricResult#getAttempted()}.
*/
public class UsesAttemptedMetrics {}
diff --git
a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesCommittedMetrics.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesCommittedMetrics.java
index c22f39747c2..6862fc43aea 100644
---
a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesCommittedMetrics.java
+++
b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/UsesCommittedMetrics.java
@@ -23,6 +23,6 @@
/**
* Category tag for validation tests which utilize {@link
org.apache.beam.sdk.metrics.Metrics}.
* Tests tagged with {@link UsesCommittedMetrics} should be run for runners
which support
- * {@link MetricResult#committed()}.
+ * {@link MetricResult#getCommitted()}.
*/
public interface UsesCommittedMetrics {}
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricResultsMatchers.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricResultsMatchers.java
index 030a75934c0..461d49e952e 100644
---
a/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricResultsMatchers.java
+++
b/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricResultsMatchers.java
@@ -57,10 +57,10 @@
return new TypeSafeMatcher<MetricResult<T>>() {
@Override
protected boolean matchesSafely(MetricResult<T> item) {
- final T metricValue = isCommitted ? item.committed() :
item.attempted();
- return Objects.equals(namespace, item.name().namespace())
- && Objects.equals(name, item.name().name())
- && item.step().contains(step)
+ final T metricValue = isCommitted ? item.getCommitted() :
item.getAttempted();
+ return Objects.equals(namespace, item.getName().getNamespace())
+ && Objects.equals(name, item.getName().getName())
+ && item.getStep().contains(step)
&& metricResultsEqual(value, metricValue);
}
@@ -77,7 +77,7 @@ public void describeTo(Description description) {
@Override
protected void describeMismatchSafely(MetricResult<T> item, Description
mismatchDescription) {
mismatchDescription.appendText("MetricResult{");
- final T metricValue = isCommitted ? item.committed() :
item.attempted();
+ final T metricValue = isCommitted ? item.getCommitted() :
item.getAttempted();
describeMetricsResultMembersMismatch(item, mismatchDescription,
namespace, name, step);
@@ -94,7 +94,7 @@ protected void describeMismatchSafely(MetricResult<T> item,
Description mismatch
private static <T> boolean metricResultsEqual(T result1, T result2) {
if (result1 instanceof GaugeResult) {
- return (((GaugeResult) result1).value()) == (((GaugeResult)
result2).value());
+ return (((GaugeResult) result1).getValue()) == (((GaugeResult)
result2).getValue());
} else {
return Objects.equals(result1, result2);
}
@@ -119,12 +119,12 @@ protected void describeMismatchSafely(MetricResult<T>
item, Description mismatch
return new TypeSafeMatcher<MetricResult<DistributionResult>>() {
@Override
protected boolean matchesSafely(MetricResult<DistributionResult> item) {
- DistributionResult metricValue = isCommitted ? item.committed() :
item.attempted();
- return Objects.equals(namespace, item.name().namespace())
- && Objects.equals(name, item.name().name())
- && item.step().contains(step)
- && Objects.equals(min, metricValue.min())
- && Objects.equals(max, metricValue.max());
+ DistributionResult metricValue = isCommitted ? item.getCommitted() :
item.getAttempted();
+ return Objects.equals(namespace, item.getName().getNamespace())
+ && Objects.equals(name, item.getName().getName())
+ && item.getStep().contains(step)
+ && Objects.equals(min, metricValue.getMin())
+ && Objects.equals(max, metricValue.getMax());
}
@Override
@@ -144,18 +144,18 @@ protected void
describeMismatchSafely(MetricResult<DistributionResult> item,
mismatchDescription.appendText("MetricResult{");
describeMetricsResultMembersMismatch(item, mismatchDescription,
namespace, name, step);
- DistributionResult metricValue = isCommitted ? item.committed() :
item.attempted();
+ DistributionResult metricValue = isCommitted ? item.getCommitted() :
item.getAttempted();
- if (!Objects.equals(min, metricValue.min())) {
+ if (!Objects.equals(min, metricValue.getMin())) {
mismatchDescription
.appendText(String.format("%sMin: ",
metricState)).appendValue(min)
- .appendText(" != ").appendValue(metricValue.min());
+ .appendText(" != ").appendValue(metricValue.getMin());
}
- if (!Objects.equals(max, metricValue.max())) {
+ if (!Objects.equals(max, metricValue.getMax())) {
mismatchDescription
.appendText(String.format("%sMax: ",
metricState)).appendValue(max)
- .appendText(" != ").appendValue(metricValue.max());
+ .appendText(" != ").appendValue(metricValue.getMax());
}
mismatchDescription.appendText("}");
@@ -169,22 +169,22 @@ protected void
describeMismatchSafely(MetricResult<DistributionResult> item,
String namespace,
String name,
String step) {
- if (!Objects.equals(namespace, item.name().namespace())) {
+ if (!Objects.equals(namespace, item.getName().getNamespace())) {
mismatchDescription
.appendText("inNamespace: ").appendValue(namespace)
- .appendText(" != ").appendValue(item.name().namespace());
+ .appendText(" != ").appendValue(item.getName().getNamespace());
}
- if (!Objects.equals(name, item.name().name())) {
+ if (!Objects.equals(name, item.getName().getName())) {
mismatchDescription
.appendText("name: ").appendValue(name)
- .appendText(" != ").appendValue(item.name().name());
+ .appendText(" != ").appendValue(item.getName().getName());
}
- if (!item.step().contains(step)) {
+ if (!item.getStep().contains(step)) {
mismatchDescription
.appendText("step: ").appendValue(step)
- .appendText(" != ").appendValue(item.step());
+ .appendText(" != ").appendValue(item.getStep());
}
}
}
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java
index bdcf8924bb2..78777eec5b2 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java
@@ -282,27 +282,27 @@ public void processElement(ProcessContext c) {
}
private static void assertCounterMetrics(MetricQueryResults metrics, boolean
isCommitted) {
- assertThat(metrics.counters(), hasItem(
+ assertThat(metrics.getCounters(), hasItem(
metricsResult(NAMESPACE, "count", "MyStep1", 3L, isCommitted)));
- assertThat(metrics.counters(), hasItem(
+ assertThat(metrics.getCounters(), hasItem(
metricsResult(NAMESPACE, "count", "MyStep2", 6L, isCommitted)));
}
private static void assertGaugeMetrics(MetricQueryResults metrics, boolean
isCommitted) {
- assertThat(metrics.gauges(), hasItem(
+ assertThat(metrics.getGauges(), hasItem(
metricsResult(NAMESPACE, "my-gauge", "MyStep2",
GaugeResult.create(12L, Instant.now()), isCommitted)));
}
private static void assertDistributionMetrics(MetricQueryResults metrics,
boolean isCommitted) {
- assertThat(metrics.distributions(), hasItem(
+ assertThat(metrics.getDistributions(), hasItem(
metricsResult(NAMESPACE, "input", "MyStep1",
DistributionResult.create(26L, 3L, 5L, 13L), isCommitted)));
- assertThat(metrics.distributions(), hasItem(
+ assertThat(metrics.getDistributions(), hasItem(
metricsResult(NAMESPACE, "input", "MyStep2",
DistributionResult.create(52L, 6L, 5L, 13L), isCommitted)));
- assertThat(metrics.distributions(), hasItem(
+ assertThat(metrics.getDistributions(), hasItem(
distributionMinMax(NAMESPACE, "bundle", "MyStep1", 10L, 40L,
isCommitted)));
}
@@ -327,13 +327,14 @@ public void testBoundedSourceMetrics() {
.queryMetrics(
MetricsFilter.builder()
.addNameFilter(
- MetricNameFilter.named(ELEMENTS_READ.namespace(),
ELEMENTS_READ.name()))
+ MetricNameFilter.named(
+ ELEMENTS_READ.getNamespace(),
ELEMENTS_READ.getName()))
.build());
- assertThat(metrics.counters(), hasItem(
+ assertThat(metrics.getCounters(), hasItem(
attemptedMetricsResult(
- ELEMENTS_READ.namespace(),
- ELEMENTS_READ.name(),
+ ELEMENTS_READ.getNamespace(),
+ ELEMENTS_READ.getName(),
"Read(BoundedCountingSource)",
1000L)));
}
@@ -355,13 +356,14 @@ public void testUnboundedSourceMetrics() {
.queryMetrics(
MetricsFilter.builder()
.addNameFilter(
- MetricNameFilter.named(ELEMENTS_READ.namespace(),
ELEMENTS_READ.name()))
+ MetricNameFilter.named(
+ ELEMENTS_READ.getNamespace(),
ELEMENTS_READ.getName()))
.build());
- assertThat(metrics.counters(), hasItem(
+ assertThat(metrics.getCounters(), hasItem(
attemptedMetricsResult(
- ELEMENTS_READ.namespace(),
- ELEMENTS_READ.name(),
+ ELEMENTS_READ.getNamespace(),
+ ELEMENTS_READ.getName(),
"Read(UnboundedCountingSource)",
1000L)));
}
diff --git
a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/PipelineRunnerTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/PipelineRunnerTest.java
index 2502476e8b3..a753ec0e0ae 100644
---
a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/PipelineRunnerTest.java
+++
b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/PipelineRunnerTest.java
@@ -102,7 +102,7 @@ public POutput expand(PBegin input) {
MetricsFilter.builder()
.addNameFilter(MetricNameFilter.inNamespace(namespace))
.build()
- ).counters(),
+ ).getCounters(),
hasItem(metricsResult(namespace, "count", "ScaleByTwo", 4L, true))
);
}
diff --git
a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java
b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java
index 7ae8f1a3ad8..235a7d9ca29 100644
---
a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java
+++
b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java
@@ -564,13 +564,13 @@ public void testUnboundedSourceWithoutBoundedWrapper() {
MetricQueryResults metrics = result.metrics().queryMetrics(
MetricsFilter.builder()
- .addNameFilter(MetricNameFilter.inNamespace(elementsRead.namespace()))
+
.addNameFilter(MetricNameFilter.inNamespace(elementsRead.getNamespace()))
.build());
- assertThat(metrics.counters(), hasItem(
+ assertThat(metrics.getCounters(), hasItem(
attemptedMetricsResult(
- elementsRead.namespace(),
- elementsRead.name(),
+ elementsRead.getNamespace(),
+ elementsRead.getName(),
"readFromKafka",
(long) numElements)));
}
@@ -768,29 +768,29 @@ public void testUnboundedSourceMetrics() {
MetricQueryResults metrics = result.metrics().queryMetrics(
MetricsFilter.builder().build());
- Iterable<MetricResult<Long>> counters = metrics.counters();
+ Iterable<MetricResult<Long>> counters = metrics.getCounters();
assertThat(counters, hasItem(attemptedMetricsResult(
- elementsRead.namespace(),
- elementsRead.name(),
+ elementsRead.getNamespace(),
+ elementsRead.getName(),
readStep,
1000L)));
assertThat(counters, hasItem(attemptedMetricsResult(
- elementsReadBySplit.namespace(),
- elementsReadBySplit.name(),
+ elementsReadBySplit.getNamespace(),
+ elementsReadBySplit.getName(),
readStep,
1000L)));
assertThat(counters, hasItem(attemptedMetricsResult(
- bytesRead.namespace(),
- bytesRead.name(),
+ bytesRead.getNamespace(),
+ bytesRead.getName(),
readStep,
12000L)));
assertThat(counters, hasItem(attemptedMetricsResult(
- bytesReadBySplit.namespace(),
- bytesReadBySplit.name(),
+ bytesReadBySplit.getNamespace(),
+ bytesReadBySplit.getName(),
readStep,
12000L)));
@@ -799,24 +799,24 @@ public void testUnboundedSourceMetrics() {
MetricsFilter.builder()
.addNameFilter(
MetricNameFilter.named(
- backlogElementsOfSplit.namespace(),
- backlogElementsOfSplit.name()))
+ backlogElementsOfSplit.getNamespace(),
+ backlogElementsOfSplit.getName()))
.build());
// since gauge values may be inconsistent in some environments assert only
on their existence.
- assertThat(backlogElementsMetrics.gauges(),
IsIterableWithSize.iterableWithSize(1));
+ assertThat(backlogElementsMetrics.getGauges(),
IsIterableWithSize.iterableWithSize(1));
MetricQueryResults backlogBytesMetrics =
result.metrics().queryMetrics(
MetricsFilter.builder()
.addNameFilter(
MetricNameFilter.named(
- backlogBytesOfSplit.namespace(),
- backlogBytesOfSplit.name()))
+ backlogBytesOfSplit.getNamespace(),
+ backlogBytesOfSplit.getName()))
.build());
// since gauge values may be inconsistent in some environments assert only
on their existence.
- assertThat(backlogBytesMetrics.gauges(),
IsIterableWithSize.iterableWithSize(1));
+ assertThat(backlogBytesMetrics.getGauges(),
IsIterableWithSize.iterableWithSize(1));
// Check checkpointMarkCommitsEnqueued metric.
MetricQueryResults commitsEnqueuedMetrics =
@@ -828,8 +828,9 @@ public void testUnboundedSourceMetrics() {
KafkaUnboundedReader.CHECKPOINT_MARK_COMMITS_ENQUEUED_METRIC))
.build());
- assertThat(commitsEnqueuedMetrics.counters(),
IsIterableWithSize.iterableWithSize(1));
-
assertThat(commitsEnqueuedMetrics.counters().iterator().next().attempted(),
greaterThan(0L));
+ assertThat(commitsEnqueuedMetrics.getCounters(),
IsIterableWithSize.iterableWithSize(1));
+ assertThat(
+ commitsEnqueuedMetrics.getCounters().iterator().next().getAttempted(),
greaterThan(0L));
}
@Test
@@ -1185,13 +1186,13 @@ public void testSinkMetrics() throws Exception {
MetricQueryResults metrics = result.metrics().queryMetrics(
MetricsFilter.builder()
-
.addNameFilter(MetricNameFilter.inNamespace(elementsWritten.namespace()))
+
.addNameFilter(MetricNameFilter.inNamespace(elementsWritten.getNamespace()))
.build());
- assertThat(metrics.counters(), hasItem(
+ assertThat(metrics.getCounters(), hasItem(
attemptedMetricsResult(
- elementsWritten.namespace(),
- elementsWritten.name(),
+ elementsWritten.getNamespace(),
+ elementsWritten.getName(),
"writeToKafka",
1000L)));
diff --git
a/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java
b/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java
index d634260dc1d..bd746ce537a 100644
---
a/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java
+++
b/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java
@@ -196,10 +196,10 @@ private long getCounterMetric(PipelineResult result,
String namespace, String na
long defaultValue) {
MetricQueryResults metrics = result.metrics().queryMetrics(
MetricsFilter.builder().addNameFilter(MetricNameFilter.named(namespace,
name)).build());
- Iterable<MetricResult<Long>> counters = metrics.counters();
+ Iterable<MetricResult<Long>> counters = metrics.getCounters();
try {
MetricResult<Long> metricResult = counters.iterator().next();
- return metricResult.attempted();
+ return metricResult.getAttempted();
} catch (NoSuchElementException e) {
LOG.error("Failed to get metric {}, from namespace {}", name, namespace);
}
@@ -214,14 +214,14 @@ private long getDistributionMetric(PipelineResult result,
String namespace, Stri
DistributionType distType, long defaultValue) {
MetricQueryResults metrics = result.metrics().queryMetrics(
MetricsFilter.builder().addNameFilter(MetricNameFilter.named(namespace,
name)).build());
- Iterable<MetricResult<DistributionResult>> distributions =
metrics.distributions();
+ Iterable<MetricResult<DistributionResult>> distributions =
metrics.getDistributions();
try {
MetricResult<DistributionResult> distributionResult =
distributions.iterator().next();
switch (distType) {
case MIN:
- return distributionResult.attempted().min();
+ return distributionResult.getAttempted().getMin();
case MAX:
- return distributionResult.attempted().max();
+ return distributionResult.getAttempted().getMax();
default:
return defaultValue;
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 84366)
Time Spent: 2h 20m (was: 2h 10m)
> Make MetricQueryResults and related classes more json-serialization friendly
> ----------------------------------------------------------------------------
>
> Key: BEAM-3892
> URL: https://issues.apache.org/jira/browse/BEAM-3892
> Project: Beam
> Issue Type: Improvement
> Components: sdk-java-core
> Reporter: Etienne Chauchot
> Assignee: Etienne Chauchot
> Priority: Major
> Time Spent: 2h 20m
> Remaining Estimate: 0h
>
> When working on this PR [https://github.com/apache/beam/pull/4548]
> MetricQueryResults needed to be serialized to be pushed to a metrics sink. As
> they were it required a custom serializer that just calls the name(),
> counter(), committed(), attempted() ... methods. MetricQueryResults are so
> close to be serializable with the default serializer, just need the accessors
> to be renamed get*, that creating DTO objects with get* methods to just call
> the non-get methods seems unnecessary.
> So just rename public accessors to get* on the experimental API
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
