You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lc...@apache.org on 2022/06/06 15:14:15 UTC
[beam] branch master updated: [Fixes #18679] Ensure that usage of metrics on a template job reports an error (#18905)
This is an automated email from the ASF dual-hosted git repository.
lcwik pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 9a7c9ce9b84 [Fixes #18679] Ensure that usage of metrics on a template job reports an error (#18905)
9a7c9ce9b84 is described below
commit 9a7c9ce9b84c0e17db0647d1652ad01e0d527eee
Author: Luke Cwik <lc...@google.com>
AuthorDate: Mon Jun 6 08:14:03 2022 -0700
[Fixes #18679] Ensure that usage of metrics on a template job reports an error (#18905)
* [Fixes #18679] Ensure that usage of metrics on a template job reports an error
Note that the usage of DataflowMetrics with a template job was throwing an NPE due to the direct access of the jobId parameter.
I also ensured that getting transform step names was accessed via a method instead of reaching into the object directly.
* fixup! reference.of.nullable warning
---
.../beam/runners/dataflow/DataflowMetrics.java | 9 +-
.../beam/runners/dataflow/DataflowPipelineJob.java | 8 +-
.../runners/dataflow/util/DataflowTemplateJob.java | 6 +
.../beam/runners/dataflow/DataflowMetricsTest.java | 125 ++++++++++++---------
4 files changed, 93 insertions(+), 55 deletions(-)
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 b58e9030a65..a4dbd04e8fc 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
@@ -40,6 +40,7 @@ import org.apache.beam.sdk.metrics.MetricResults;
import org.apache.beam.sdk.metrics.MetricsFilter;
import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Objects;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.BiMap;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.slf4j.Logger;
@@ -114,7 +115,7 @@ class DataflowMetrics extends MetricResults {
// Metric results have been cached after the job ran.
return cachedMetricResults;
}
- JobMetrics result = dataflowClient.getJobMetrics(dataflowPipelineJob.jobId);
+ JobMetrics result = dataflowClient.getJobMetrics(dataflowPipelineJob.getJobId());
if (dataflowPipelineJob.getState().isTerminal()) {
// Add current query result to the cache.
cachedMetricResults = result;
@@ -276,13 +277,15 @@ class DataflowMetrics extends MetricResults {
private @Nullable String getNonPortableUserStepName(String internalStepName) {
// If we can't translate internal step names to user step names, we just skip them altogether.
- if (dataflowPipelineJob.transformStepNames == null) {
+ BiMap<AppliedPTransform<?, ?, ?>, String> transformStepNames =
+ dataflowPipelineJob.getTransformStepNames();
+ if (transformStepNames == null) {
return null;
}
@Nullable
AppliedPTransform<?, ?, ?> appliedPTransform =
- dataflowPipelineJob.transformStepNames.inverse().get(internalStepName);
+ transformStepNames.inverse().get(internalStepName);
if (appliedPTransform == null) {
return null;
}
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineJob.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineJob.java
index 174bb6acd72..35a6318cf05 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineJob.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineJob.java
@@ -61,7 +61,7 @@ public class DataflowPipelineJob implements PipelineResult {
private static final Logger LOG = LoggerFactory.getLogger(DataflowPipelineJob.class);
/** The id for the job. */
- protected String jobId;
+ private final String jobId;
/** The {@link DataflowPipelineOptions} for the job. */
private final DataflowPipelineOptions dataflowOptions;
@@ -84,7 +84,7 @@ public class DataflowPipelineJob implements PipelineResult {
/** The job that replaced this one or {@code null} if the job has not been replaced. */
private @Nullable DataflowPipelineJob replacedByJob = null;
- protected BiMap<AppliedPTransform<?, ?, ?>, String> transformStepNames;
+ private final BiMap<AppliedPTransform<?, ?, ?>, String> transformStepNames;
/** The latest timestamp up to which job messages have been retrieved. */
private long lastTimestamp = Long.MIN_VALUE;
@@ -181,6 +181,10 @@ public class DataflowPipelineJob implements PipelineResult {
return dataflowOptions.getRegion();
}
+ protected @Nullable BiMap<AppliedPTransform<?, ?, ?>, String> getTransformStepNames() {
+ return transformStepNames;
+ }
+
/**
* Returns a new {@link DataflowPipelineJob} for the job that replaced this one, if applicable.
*
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/DataflowTemplateJob.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/DataflowTemplateJob.java
index 1047916fea6..ea26203bde4 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/DataflowTemplateJob.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/DataflowTemplateJob.java
@@ -20,6 +20,7 @@ package org.apache.beam.runners.dataflow.util;
import com.google.api.client.util.NanoClock;
import com.google.api.client.util.Sleeper;
import org.apache.beam.runners.dataflow.DataflowPipelineJob;
+import org.apache.beam.sdk.metrics.MetricResults;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.joda.time.Duration;
@@ -50,6 +51,11 @@ public class DataflowTemplateJob extends DataflowPipelineJob {
throw new UnsupportedOperationException(ERROR);
}
+ @Override
+ public MetricResults metrics() {
+ throw new UnsupportedOperationException(ERROR);
+ }
+
@VisibleForTesting
@Nullable
State waitUntilFinish(
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 2ece1f8436e..7f2b239098e 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
@@ -25,6 +25,7 @@ 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.assertThrows;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
@@ -40,13 +41,16 @@ import com.google.api.services.dataflow.model.MetricUpdate;
import java.io.IOException;
import java.math.BigDecimal;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
+import org.apache.beam.runners.dataflow.util.DataflowTemplateJob;
import org.apache.beam.sdk.PipelineResult.State;
import org.apache.beam.sdk.extensions.gcp.auth.TestCredential;
import org.apache.beam.sdk.extensions.gcp.storage.NoopPathValidator;
import org.apache.beam.sdk.metrics.DistributionResult;
import org.apache.beam.sdk.metrics.MetricQueryResults;
+import org.apache.beam.sdk.metrics.MetricsFilter;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.BiMap;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.HashBiMap;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
@@ -100,7 +104,7 @@ public class DataflowMetricsTest {
when(options.isStreaming()).thenReturn(false);
when(job.getDataflowOptions()).thenReturn(options);
when(job.getState()).thenReturn(State.RUNNING);
- job.jobId = JOB_ID;
+ when(job.getJobId()).thenReturn(JOB_ID);
JobMetrics jobMetrics = new JobMetrics();
jobMetrics.setMetrics(null /* this is how the APIs represent empty metrics */);
@@ -123,7 +127,7 @@ public class DataflowMetricsTest {
when(options.isStreaming()).thenReturn(false);
when(job.getDataflowOptions()).thenReturn(options);
when(job.getState()).thenReturn(State.DONE);
- job.jobId = JOB_ID;
+ when(job.getJobId()).thenReturn(JOB_ID);
JobMetrics jobMetrics = new JobMetrics();
jobMetrics.setMetrics(ImmutableList.of());
@@ -182,18 +186,19 @@ public class DataflowMetricsTest {
@Test
public void testSingleCounterUpdates() throws IOException {
+ AppliedPTransform<?, ?, ?> myStep = mock(AppliedPTransform.class);
+ when(myStep.getFullName()).thenReturn("myStepName");
+ BiMap<AppliedPTransform<?, ?, ?>, String> transformStepNames = HashBiMap.create();
+ transformStepNames.put(myStep, "s2");
+
JobMetrics jobMetrics = new JobMetrics();
DataflowPipelineJob job = mock(DataflowPipelineJob.class);
DataflowPipelineOptions options = mock(DataflowPipelineOptions.class);
when(options.isStreaming()).thenReturn(false);
when(job.getDataflowOptions()).thenReturn(options);
when(job.getState()).thenReturn(State.RUNNING);
- job.jobId = JOB_ID;
-
- AppliedPTransform<?, ?, ?> myStep = mock(AppliedPTransform.class);
- when(myStep.getFullName()).thenReturn("myStepName");
- job.transformStepNames = HashBiMap.create();
- job.transformStepNames.put(myStep, "s2");
+ when(job.getJobId()).thenReturn(JOB_ID);
+ when(job.getTransformStepNames()).thenReturn(transformStepNames);
MetricUpdate update = new MetricUpdate();
long stepValue = 1234L;
@@ -223,6 +228,11 @@ public class DataflowMetricsTest {
@Test
public void testIgnoreDistributionButGetCounterUpdates() throws IOException {
+ AppliedPTransform<?, ?, ?> myStep = mock(AppliedPTransform.class);
+ when(myStep.getFullName()).thenReturn("myStepName");
+ BiMap<AppliedPTransform<?, ?, ?>, String> transformStepNames = HashBiMap.create();
+ transformStepNames.put(myStep, "s2");
+
JobMetrics jobMetrics = new JobMetrics();
DataflowClient dataflowClient = mock(DataflowClient.class);
when(dataflowClient.getJobMetrics(JOB_ID)).thenReturn(jobMetrics);
@@ -231,12 +241,8 @@ public class DataflowMetricsTest {
when(options.isStreaming()).thenReturn(false);
when(job.getDataflowOptions()).thenReturn(options);
when(job.getState()).thenReturn(State.RUNNING);
- job.jobId = JOB_ID;
-
- AppliedPTransform<?, ?, ?> myStep = mock(AppliedPTransform.class);
- when(myStep.getFullName()).thenReturn("myStepName");
- job.transformStepNames = HashBiMap.create();
- job.transformStepNames.put(myStep, "s2");
+ when(job.getJobId()).thenReturn(JOB_ID);
+ when(job.getTransformStepNames()).thenReturn(transformStepNames);
// The parser relies on the fact that one tentative and one committed metric update exist in
// the job metrics results.
@@ -261,6 +267,11 @@ public class DataflowMetricsTest {
@Test
public void testDistributionUpdates() throws IOException {
+ AppliedPTransform<?, ?, ?> myStep2 = mock(AppliedPTransform.class);
+ when(myStep2.getFullName()).thenReturn("myStepName");
+ BiMap<AppliedPTransform<?, ?, ?>, String> transformStepNames = HashBiMap.create();
+ transformStepNames.put(myStep2, "s2");
+
JobMetrics jobMetrics = new JobMetrics();
DataflowClient dataflowClient = mock(DataflowClient.class);
when(dataflowClient.getJobMetrics(JOB_ID)).thenReturn(jobMetrics);
@@ -269,12 +280,8 @@ public class DataflowMetricsTest {
when(options.isStreaming()).thenReturn(false);
when(job.getDataflowOptions()).thenReturn(options);
when(job.getState()).thenReturn(State.RUNNING);
- job.jobId = JOB_ID;
-
- AppliedPTransform<?, ?, ?> myStep2 = mock(AppliedPTransform.class);
- when(myStep2.getFullName()).thenReturn("myStepName");
- job.transformStepNames = HashBiMap.create();
- job.transformStepNames.put(myStep2, "s2");
+ when(job.getJobId()).thenReturn(JOB_ID);
+ when(job.getTransformStepNames()).thenReturn(transformStepNames);
// The parser relies on the fact that one tentative and one committed metric update exist in
// the job metrics results.
@@ -307,6 +314,11 @@ public class DataflowMetricsTest {
@Test
public void testDistributionUpdatesStreaming() throws IOException {
+ AppliedPTransform<?, ?, ?> myStep2 = mock(AppliedPTransform.class);
+ when(myStep2.getFullName()).thenReturn("myStepName");
+ BiMap<AppliedPTransform<?, ?, ?>, String> transformStepNames = HashBiMap.create();
+ transformStepNames.put(myStep2, "s2");
+
JobMetrics jobMetrics = new JobMetrics();
DataflowClient dataflowClient = mock(DataflowClient.class);
when(dataflowClient.getJobMetrics(JOB_ID)).thenReturn(jobMetrics);
@@ -315,12 +327,8 @@ public class DataflowMetricsTest {
when(options.isStreaming()).thenReturn(true);
when(job.getDataflowOptions()).thenReturn(options);
when(job.getState()).thenReturn(State.RUNNING);
- job.jobId = JOB_ID;
-
- AppliedPTransform<?, ?, ?> myStep2 = mock(AppliedPTransform.class);
- when(myStep2.getFullName()).thenReturn("myStepName");
- job.transformStepNames = HashBiMap.create();
- job.transformStepNames.put(myStep2, "s2");
+ when(job.getJobId()).thenReturn(JOB_ID);
+ when(job.getTransformStepNames()).thenReturn(transformStepNames);
// The parser relies on the fact that one tentative and one committed metric update exist in
// the job metrics results.
@@ -355,6 +363,17 @@ public class DataflowMetricsTest {
@Test
public void testMultipleCounterUpdates() throws IOException {
+ AppliedPTransform<?, ?, ?> myStep2 = mock(AppliedPTransform.class);
+ when(myStep2.getFullName()).thenReturn("myStepName");
+ BiMap<AppliedPTransform<?, ?, ?>, String> transformStepNames = HashBiMap.create();
+ transformStepNames.put(myStep2, "s2");
+ AppliedPTransform<?, ?, ?> myStep3 = mock(AppliedPTransform.class);
+ when(myStep3.getFullName()).thenReturn("myStepName3");
+ transformStepNames.put(myStep3, "s3");
+ AppliedPTransform<?, ?, ?> myStep4 = mock(AppliedPTransform.class);
+ when(myStep4.getFullName()).thenReturn("myStepName4");
+ transformStepNames.put(myStep4, "s4");
+
JobMetrics jobMetrics = new JobMetrics();
DataflowClient dataflowClient = mock(DataflowClient.class);
when(dataflowClient.getJobMetrics(JOB_ID)).thenReturn(jobMetrics);
@@ -363,18 +382,8 @@ public class DataflowMetricsTest {
when(options.isStreaming()).thenReturn(false);
when(job.getDataflowOptions()).thenReturn(options);
when(job.getState()).thenReturn(State.RUNNING);
- job.jobId = JOB_ID;
-
- AppliedPTransform<?, ?, ?> myStep2 = mock(AppliedPTransform.class);
- when(myStep2.getFullName()).thenReturn("myStepName");
- job.transformStepNames = HashBiMap.create();
- job.transformStepNames.put(myStep2, "s2");
- AppliedPTransform<?, ?, ?> myStep3 = mock(AppliedPTransform.class);
- when(myStep3.getFullName()).thenReturn("myStepName3");
- job.transformStepNames.put(myStep3, "s3");
- AppliedPTransform<?, ?, ?> myStep4 = mock(AppliedPTransform.class);
- when(myStep4.getFullName()).thenReturn("myStepName4");
- job.transformStepNames.put(myStep4, "s4");
+ when(job.getJobId()).thenReturn(JOB_ID);
+ when(job.getTransformStepNames()).thenReturn(transformStepNames);
// The parser relies on the fact that one tentative and one committed metric update exist in
// the job metrics results.
@@ -408,6 +417,17 @@ public class DataflowMetricsTest {
@Test
public void testMultipleCounterUpdatesStreaming() throws IOException {
+ AppliedPTransform<?, ?, ?> myStep2 = mock(AppliedPTransform.class);
+ when(myStep2.getFullName()).thenReturn("myStepName");
+ BiMap<AppliedPTransform<?, ?, ?>, String> transformStepNames = HashBiMap.create();
+ transformStepNames.put(myStep2, "s2");
+ AppliedPTransform<?, ?, ?> myStep3 = mock(AppliedPTransform.class);
+ when(myStep3.getFullName()).thenReturn("myStepName3");
+ transformStepNames.put(myStep3, "s3");
+ AppliedPTransform<?, ?, ?> myStep4 = mock(AppliedPTransform.class);
+ when(myStep4.getFullName()).thenReturn("myStepName4");
+ transformStepNames.put(myStep4, "s4");
+
JobMetrics jobMetrics = new JobMetrics();
DataflowClient dataflowClient = mock(DataflowClient.class);
when(dataflowClient.getJobMetrics(JOB_ID)).thenReturn(jobMetrics);
@@ -416,18 +436,8 @@ public class DataflowMetricsTest {
when(options.isStreaming()).thenReturn(true);
when(job.getDataflowOptions()).thenReturn(options);
when(job.getState()).thenReturn(State.RUNNING);
- job.jobId = JOB_ID;
-
- AppliedPTransform<?, ?, ?> myStep2 = mock(AppliedPTransform.class);
- when(myStep2.getFullName()).thenReturn("myStepName");
- job.transformStepNames = HashBiMap.create();
- job.transformStepNames.put(myStep2, "s2");
- AppliedPTransform<?, ?, ?> myStep3 = mock(AppliedPTransform.class);
- when(myStep3.getFullName()).thenReturn("myStepName3");
- job.transformStepNames.put(myStep3, "s3");
- AppliedPTransform<?, ?, ?> myStep4 = mock(AppliedPTransform.class);
- when(myStep4.getFullName()).thenReturn("myStepName4");
- job.transformStepNames.put(myStep4, "s4");
+ when(job.getJobId()).thenReturn(JOB_ID);
+ when(job.getTransformStepNames()).thenReturn(transformStepNames);
// The parser relies on the fact that one tentative and one committed metric update exist in
// the job metrics results.
@@ -459,4 +469,19 @@ public class DataflowMetricsTest {
attemptedMetricsResult("otherNamespace", "otherCounter", "myStepName3", 12L),
attemptedMetricsResult("otherNamespace", "counterName", "myStepName4", 1200L)));
}
+
+ @Test
+ public void testTemplateJobMetricsThrowsUsefulError() throws Exception {
+ DataflowClient dataflowClient = mock(DataflowClient.class);
+ DataflowMetrics metrics = new DataflowMetrics(new DataflowTemplateJob(), dataflowClient);
+
+ assertThrows(
+ "The result of template creation should not be used.",
+ UnsupportedOperationException.class,
+ () -> metrics.allMetrics());
+ assertThrows(
+ "The result of template creation should not be used.",
+ UnsupportedOperationException.class,
+ () -> metrics.queryMetrics(MetricsFilter.builder().build()));
+ }
}