[
https://issues.apache.org/jira/browse/BEAM-3803?focusedWorklogId=79249&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-79249
]
ASF GitHub Bot logged work on BEAM-3803:
----------------------------------------
Author: ASF GitHub Bot
Created on: 10/Mar/18 20:11
Start Date: 10/Mar/18 20:11
Worklog Time Spent: 10m
Work Description: kennknowles closed pull request #4841: BEAM-3803:
Dataflow runner implements metrics contract
URL: https://github.com/apache/beam/pull/4841
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/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 0983674b983..376befa77f4 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
@@ -153,9 +153,10 @@ public void addMetricResult(
metricKey.metricName(),
metricKey.stepName(),
isStreamingJob ? null : value, // Committed
- isStreamingJob ? value : null)); // Attempted
+ value)); // Attempted
/* In Dataflow streaming jobs, only ATTEMPTED metrics are available.
- * In Dataflow batch jobs, only COMMITTED metrics are available.
+ * In Dataflow batch jobs, only COMMITTED metrics are available, but
+ * we must provide ATTEMPTED, so we use COMMITTED as a good
approximation.
* Reporting the appropriate metric depending on whether it's a
batch/streaming job.
*/
} else if (committed.getScalar() != null && attempted.getScalar() !=
null) {
@@ -166,9 +167,10 @@ public void addMetricResult(
metricKey.metricName(),
metricKey.stepName(),
isStreamingJob ? null : value, // Committed
- isStreamingJob ? value : null)); // Attempted
+ value)); // Attempted
/* In Dataflow streaming jobs, only ATTEMPTED metrics are available.
- * In Dataflow batch jobs, only COMMITTED metrics are available.
+ * In Dataflow batch jobs, only COMMITTED metrics are available, but
+ * we must provide ATTEMPTED, so we use COMMITTED as a good
approximation.
* Reporting the appropriate metric depending on whether it's a
batch/streaming job.
*/
} else {
@@ -350,10 +352,18 @@ public static MetricQueryResults create(
public abstract MetricName name();
public abstract String step();
@Nullable
- public abstract T committed();
- @Nullable
+ protected abstract T committedInternal();
public abstract T attempted();
+ public T committed() {
+ T committed = committedInternal();
+ if (committed == null) {
+ throw new UnsupportedOperationException("This runner does not
currently support committed"
+ + " metrics results. Please use 'attempted' instead.");
+ }
+ return committed;
+ }
+
public static <T> MetricResult<T> create(MetricName name, String scope,
T committed, T attempted) {
return new AutoValue_DataflowMetrics_DataflowMetricResult<>(
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 baf02114179..7aa291065aa 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
@@ -22,8 +22,10 @@
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -208,7 +210,7 @@ public void testSingleCounterUpdates() throws IOException {
DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient);
MetricQueryResults result = dataflowMetrics.queryMetrics(null);
assertThat(result.counters(), containsInAnyOrder(
- attemptedMetricsResult("counterNamespace", "counterName",
"myStepName", (Long) null)));
+ attemptedMetricsResult("counterNamespace", "counterName",
"myStepName", 1234L)));
assertThat(result.counters(), containsInAnyOrder(
committedMetricsResult("counterNamespace", "counterName",
"myStepName", 1234L)));
}
@@ -241,7 +243,7 @@ public void testIgnoreDistributionButGetCounterUpdates()
throws IOException {
DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient);
MetricQueryResults result = dataflowMetrics.queryMetrics(null);
assertThat(result.counters(), containsInAnyOrder(
- attemptedMetricsResult("counterNamespace", "counterName",
"myStepName", (Long) null)));
+ attemptedMetricsResult("counterNamespace", "counterName",
"myStepName", 1233L)));
assertThat(result.counters(), containsInAnyOrder(
committedMetricsResult("counterNamespace", "counterName",
"myStepName", 1233L)));
}
@@ -275,7 +277,7 @@ public void testDistributionUpdates() throws IOException {
MetricQueryResults result = dataflowMetrics.queryMetrics(null);
assertThat(result.distributions(), contains(
attemptedMetricsResult("distributionNamespace", "distributionName",
"myStepName",
- (DistributionResult) null)));
+ DistributionResult.create(18, 2, 2, 16))));
assertThat(result.distributions(), contains(
committedMetricsResult("distributionNamespace", "distributionName",
"myStepName",
DistributionResult.create(18, 2, 2, 16))));
@@ -308,9 +310,14 @@ public void testDistributionUpdatesStreaming() throws
IOException {
DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient);
MetricQueryResults result = dataflowMetrics.queryMetrics(null);
- assertThat(result.distributions(), contains(
- committedMetricsResult("distributionNamespace", "distributionName",
"myStepName",
- (DistributionResult) null)));
+ try {
+ result.distributions().iterator().next().committed();
+ 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(
attemptedMetricsResult("distributionNamespace", "distributionName",
"myStepName",
DistributionResult.create(18, 2, 2, 16))));
@@ -356,9 +363,9 @@ public void testMultipleCounterUpdates() throws IOException
{
DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient);
MetricQueryResults result = dataflowMetrics.queryMetrics(null);
assertThat(result.counters(), containsInAnyOrder(
- attemptedMetricsResult("counterNamespace", "counterName",
"myStepName", (Long) null),
- attemptedMetricsResult("otherNamespace", "otherCounter",
"myStepName3", (Long) null),
- attemptedMetricsResult("otherNamespace", "counterName", "myStepName4",
(Long) null)));
+ attemptedMetricsResult("counterNamespace", "counterName",
"myStepName", 1233L),
+ attemptedMetricsResult("otherNamespace", "otherCounter",
"myStepName3", 12L),
+ attemptedMetricsResult("otherNamespace", "counterName", "myStepName4",
1200L)));
assertThat(result.counters(), containsInAnyOrder(
committedMetricsResult("counterNamespace", "counterName",
"myStepName", 1233L),
committedMetricsResult("otherNamespace", "otherCounter",
"myStepName3", 12L),
@@ -400,10 +407,14 @@ public void testMultipleCounterUpdatesStreaming() throws
IOException {
DataflowMetrics dataflowMetrics = new DataflowMetrics(job, dataflowClient);
MetricQueryResults result = dataflowMetrics.queryMetrics(null);
- assertThat(result.counters(), containsInAnyOrder(
- committedMetricsResult("counterNamespace", "counterName",
"myStepName", (Long) null),
- committedMetricsResult("otherNamespace", "otherCounter",
"myStepName3", (Long) null),
- committedMetricsResult("otherNamespace", "counterName", "myStepName4",
(Long) null)));
+ try {
+ result.counters().iterator().next().committed();
+ 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(
attemptedMetricsResult("counterNamespace", "counterName",
"myStepName", 1233L),
attemptedMetricsResult("otherNamespace", "otherCounter",
"myStepName3", 12L),
----------------------------------------------------------------
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: 79249)
Time Spent: 4h 40m (was: 4.5h)
> Dataflow runner should handle metrics per the spec
> --------------------------------------------------
>
> Key: BEAM-3803
> URL: https://issues.apache.org/jira/browse/BEAM-3803
> Project: Beam
> Issue Type: Bug
> Components: runner-dataflow
> Reporter: Andrew Pilloud
> Assignee: Andrew Pilloud
> Priority: Major
> Labels: nexmark
> Time Spent: 4h 40m
> Remaining Estimate: 0h
>
> The dataflow runner only supports committed metrics for batch jobs and
> attempted metrics for streaming jobs. It should always support attempted
> metrics and throw an UnsupportedOperationException when the metrics are
> missing.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
