This is an automated email from the ASF dual-hosted git repository.
Amar3tto pushed a commit to branch icerbergio-benchmark
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/icerbergio-benchmark by this
push:
new f16991fd1cf Refactor DefaultPipelineLauncher
f16991fd1cf is described below
commit f16991fd1cfbb64a983e8aea0ef3e809d45f301a
Author: Vitaly Terentyev <[email protected]>
AuthorDate: Tue Jul 7 15:58:46 2026 +0400
Refactor DefaultPipelineLauncher
---
.../common/dataflow/DefaultPipelineLauncher.java | 45 ++++++++++++++--------
.../org/apache/beam/it/common/IOLoadTestBase.java | 18 ++-------
.../org/apache/beam/it/common/LoadTestBase.java | 1 -
.../bigquery/BigQueryResourceManagerTest.java | 4 +-
.../bigquery/BigQueryResourceManagerUtilsTest.java | 3 +-
.../dataflow/AbstractPipelineLauncherTest.java | 3 +-
.../dataflow/DefaultPipelineLauncherTest.java | 10 ++---
7 files changed, 39 insertions(+), 45 deletions(-)
diff --git
a/it/common/src/main/java/org/apache/beam/it/common/dataflow/DefaultPipelineLauncher.java
b/it/common/src/main/java/org/apache/beam/it/common/dataflow/DefaultPipelineLauncher.java
index a872868c12b..32a4eb6eb1b 100644
---
a/it/common/src/main/java/org/apache/beam/it/common/dataflow/DefaultPipelineLauncher.java
+++
b/it/common/src/main/java/org/apache/beam/it/common/dataflow/DefaultPipelineLauncher.java
@@ -18,7 +18,6 @@
package org.apache.beam.it.common.dataflow;
import static org.apache.beam.it.common.logging.LogStrings.formatForLogging;
-import static
org.apache.beam.sdk.testing.TestPipeline.PROPERTY_BEAM_TEST_PIPELINE_OPTIONS;
import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -41,7 +40,6 @@ import java.util.regex.Pattern;
import java.util.stream.StreamSupport;
import org.apache.beam.it.common.PipelineLauncher;
import org.apache.beam.it.common.utils.PipelineUtils;
-import org.apache.beam.it.common.IOLoadTestBase;
import org.apache.beam.runners.dataflow.DataflowPipelineJob;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.metrics.DistributionResult;
@@ -73,6 +71,8 @@ public class DefaultPipelineLauncher extends
AbstractPipelineLauncher {
private static final String READ_PIPELINE_NAME_OVERWRITE =
"readPipelineNameOverride";
private static final String WRITE_PIPELINE_NAME_OVERWRITE =
"writePipelineNameOverride";
private static final Pattern JOB_ID_PATTERN = Pattern.compile("Submitted
job: (\\S+)");
+ /** Namespace for Beam provided pipeline metrics (set up by Metrics
transform). */
+ public static final String BEAM_METRICS_NAMESPACE = "BEAM_METRICS";
// For unsupported runners (other than dataflow), implement launcher methods
by operating with
// PipelineResult.
@@ -101,6 +101,18 @@ public class DefaultPipelineLauncher extends
AbstractPipelineLauncher {
.put(PipelineResult.State.UNRECOGNIZED, JobState.UNKNOWN)
.build();
+ // To make PipelineLauncher.getMetric work in a unified way for both runner
provided metrics and
+ // pipeline defined
+ // metrics, here we wrap Beam provided metrics as a pre-defined metrics name
+ // [name_space:metric_type:metric_name
+ // which will be recognized by getMetric method
+ public enum PipelineMetricsType {
+ COUNTER,
+ STARTTIME,
+ ENDTIME,
+ RUNTIME,
+ }
+
private DefaultPipelineLauncher(Builder builder) {
super(
new Dataflow(
@@ -169,7 +181,7 @@ public class DefaultPipelineLauncher extends
AbstractPipelineLauncher {
resultCount <= 1,
"More than one metric result matches name: %s in namespace %s. Metric
results count: %s",
name,
- IOLoadTestBase.BEAM_METRICS_NAMESPACE,
+ BEAM_METRICS_NAMESPACE,
resultCount);
}
@@ -181,14 +193,14 @@ public class DefaultPipelineLauncher extends
AbstractPipelineLauncher {
.queryMetrics(
MetricsFilter.builder()
.addNameFilter(
-
MetricNameFilter.named(IOLoadTestBase.BEAM_METRICS_NAMESPACE, metricName))
+ MetricNameFilter.named(BEAM_METRICS_NAMESPACE,
metricName))
.build());
return metrics.getDistributions();
}
/** Pull Beam pipeline defined metrics given the jobId. */
public Long getBeamMetric(
- String jobId, IOLoadTestBase.PipelineMetricsType metricType, String
metricName) {
+ String jobId, PipelineMetricsType metricType, String metricName) {
PipelineResult pipelineResult =
MANAGED_JOBS.getOrDefault(jobId, UNMANAGED_JOBS.getOrDefault(jobId,
null));
if (pipelineResult != null) {
@@ -198,7 +210,7 @@ public class DefaultPipelineLauncher extends
AbstractPipelineLauncher {
.queryMetrics(
MetricsFilter.builder()
.addNameFilter(
-
MetricNameFilter.named(IOLoadTestBase.BEAM_METRICS_NAMESPACE, metricName))
+ MetricNameFilter.named(BEAM_METRICS_NAMESPACE,
metricName))
.build());
switch (metricType) {
@@ -212,7 +224,7 @@ public class DefaultPipelineLauncher extends
AbstractPipelineLauncher {
LOG.error(
"Failed to get metric {}, from namespace {}",
metricName,
- IOLoadTestBase.BEAM_METRICS_NAMESPACE);
+ BEAM_METRICS_NAMESPACE);
}
return UNKNOWN_METRIC_VALUE;
case STARTTIME:
@@ -230,9 +242,9 @@ public class DefaultPipelineLauncher extends
AbstractPipelineLauncher {
.map(element ->
Objects.requireNonNull(element.getAttempted()).getMax())
.max(Long::compareTo)
.orElse(UNKNOWN_METRIC_VALUE);
- if (metricType == IOLoadTestBase.PipelineMetricsType.STARTTIME) {
+ if (metricType == PipelineMetricsType.STARTTIME) {
return lowestMin;
- } else if (metricType == IOLoadTestBase.PipelineMetricsType.ENDTIME)
{
+ } else if (metricType == PipelineMetricsType.ENDTIME) {
return greatestMax;
} else {
if (lowestMin != UNKNOWN_METRIC_VALUE && greatestMax !=
UNKNOWN_METRIC_VALUE) {
@@ -254,15 +266,15 @@ public class DefaultPipelineLauncher extends
AbstractPipelineLauncher {
@Override
public Double getMetric(String project, String region, String jobId, String
metricName)
throws IOException {
- if (metricName.startsWith(IOLoadTestBase.BEAM_METRICS_NAMESPACE)) {
+ if (metricName.startsWith(BEAM_METRICS_NAMESPACE)) {
String[] nameSpacedMetrics = metricName.split(":", 3);
Preconditions.checkState(
nameSpacedMetrics.length == 3,
String.format(
"Invalid Beam metrics name: %s, expected:
'%s:metric_type:metric_name'",
- metricName, IOLoadTestBase.BEAM_METRICS_NAMESPACE));
- IOLoadTestBase.PipelineMetricsType metricType =
- IOLoadTestBase.PipelineMetricsType.valueOf(nameSpacedMetrics[1]);
+ metricName, BEAM_METRICS_NAMESPACE));
+ PipelineMetricsType metricType =
+ PipelineMetricsType.valueOf(nameSpacedMetrics[1]);
// Pipeline defined metrics are long values. Have to cast to double that
is what the base
// class defined.
@@ -437,16 +449,15 @@ public class DefaultPipelineLauncher extends
AbstractPipelineLauncher {
// add pipeline options from beamTestPipelineOptions system property to
preserve the
// pipeline options already set in TestPipeline.
@Nullable
- String beamTestPipelineOptions =
System.getProperty(PROPERTY_BEAM_TEST_PIPELINE_OPTIONS);
+ String beamTestPipelineOptions =
System.getProperty(org.apache.beam.sdk.testing.TestPipeline.PROPERTY_BEAM_TEST_PIPELINE_OPTIONS);
if (!Strings.isNullOrEmpty(beamTestPipelineOptions)) {
try {
additionalOptions.addAll(MAPPER.readValue(beamTestPipelineOptions,
List.class));
} catch (IOException e) {
throw new RuntimeException(
"Unable to instantiate test options from system property "
- + PROPERTY_BEAM_TEST_PIPELINE_OPTIONS
- + ":"
- + System.getProperty(PROPERTY_BEAM_TEST_PIPELINE_OPTIONS),
+ +
org.apache.beam.sdk.testing.TestPipeline.PROPERTY_BEAM_TEST_PIPELINE_OPTIONS
+ + ":" + beamTestPipelineOptions,
e);
}
}
diff --git
a/it/common/src/test/java/org/apache/beam/it/common/IOLoadTestBase.java
b/it/common/src/test/java/org/apache/beam/it/common/IOLoadTestBase.java
index fb9feb6663b..78ec2af36d7 100644
--- a/it/common/src/test/java/org/apache/beam/it/common/IOLoadTestBase.java
+++ b/it/common/src/test/java/org/apache/beam/it/common/IOLoadTestBase.java
@@ -25,6 +25,7 @@ import java.util.Map;
import java.util.UUID;
import org.apache.beam.it.common.dataflow.DefaultPipelineLauncher;
+import
org.apache.beam.it.common.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.metrics.Metrics;
import org.apache.beam.sdk.testutils.NamedTestResult;
@@ -38,6 +39,8 @@ import org.junit.runners.JUnit4;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static
org.apache.beam.it.common.dataflow.DefaultPipelineLauncher.BEAM_METRICS_NAMESPACE;
+
/** Base class for IO Load tests. */
@RunWith(JUnit4.class)
@SuppressWarnings({
@@ -89,21 +92,6 @@ public class IOLoadTestBase extends LoadTestBase {
}
}
- // To make PipelineLauncher.getMetric work in a unified way for both runner
provided metrics and
- // pipeline defined
- // metrics, here we wrap Beam provided metrics as a pre-defined metrics name
- // [name_space:metric_type:metric_name
- // which will be recognized by getMetric method
- public enum PipelineMetricsType {
- COUNTER,
- STARTTIME,
- ENDTIME,
- RUNTIME,
- }
-
- /** Namespace for Beam provided pipeline metrics (set up by Metrics
transform). */
- public static final String BEAM_METRICS_NAMESPACE = "BEAM_METRICS";
-
/** Given a metrics name, return Beam metrics name. */
public static String getBeamMetricsName(PipelineMetricsType metricstype,
String metricsName) {
return BEAM_METRICS_NAMESPACE + ":" + metricstype + ":" + metricsName;
diff --git
a/it/common/src/test/java/org/apache/beam/it/common/LoadTestBase.java
b/it/common/src/test/java/org/apache/beam/it/common/LoadTestBase.java
index 7d7adbbae0a..bb0d2ab72bd 100644
--- a/it/common/src/test/java/org/apache/beam/it/common/LoadTestBase.java
+++ b/it/common/src/test/java/org/apache/beam/it/common/LoadTestBase.java
@@ -43,7 +43,6 @@ import java.util.regex.Pattern;
import org.apache.beam.it.common.PipelineLauncher.LaunchInfo;
import org.apache.beam.it.common.bigquery.BigQueryResourceManager;
-import org.apache.beam.it.common.dataflow.AbstractPipelineLauncher;
import org.apache.beam.it.common.monitoring.MonitoringClient;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects;
import org.checkerframework.checker.nullness.qual.Nullable;
diff --git
a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerTest.java
b/it/common/src/test/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerTest.java
similarity index 98%
rename from
it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerTest.java
rename to
it/common/src/test/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerTest.java
index 3a633f00c6f..62aac93b4ee 100644
---
a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerTest.java
+++
b/it/common/src/test/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerTest.java
@@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.it.gcp.bigquery;
+package org.apache.beam.it.common.bigquery;
import static com.google.common.truth.Truth.assertThat;
import static org.junit.Assert.assertThrows;
@@ -41,8 +41,6 @@ import com.google.cloud.bigquery.Table;
import com.google.cloud.bigquery.TableId;
import com.google.cloud.bigquery.TableInfo;
import com.google.cloud.bigquery.TimePartitioning;
-import org.apache.beam.it.common.bigquery.BigQueryResourceManager;
-import org.apache.beam.it.common.bigquery.BigQueryResourceManagerException;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
import org.junit.Before;
diff --git
a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerUtilsTest.java
b/it/common/src/test/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerUtilsTest.java
similarity index 94%
rename from
it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerUtilsTest.java
rename to
it/common/src/test/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerUtilsTest.java
index 7248fed0e87..eaff8074072 100644
---
a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerUtilsTest.java
+++
b/it/common/src/test/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerUtilsTest.java
@@ -15,14 +15,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.it.gcp.bigquery;
+package org.apache.beam.it.common.bigquery;
import static
org.apache.beam.it.common.bigquery.BigQueryResourceManagerUtils.checkValidTableId;
import static org.junit.Assert.assertThrows;
import java.util.Arrays;
-import org.apache.beam.it.common.bigquery.BigQueryResourceManagerUtils;
import org.junit.Test;
/** Unit tests for {@link BigQueryResourceManagerUtils}. */
diff --git
a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/AbstractPipelineLauncherTest.java
b/it/common/src/test/java/org/apache/beam/it/common/dataflow/AbstractPipelineLauncherTest.java
similarity index 98%
rename from
it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/AbstractPipelineLauncherTest.java
rename to
it/common/src/test/java/org/apache/beam/it/common/dataflow/AbstractPipelineLauncherTest.java
index df1c5cea640..72d5514a6e3 100644
---
a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/AbstractPipelineLauncherTest.java
+++
b/it/common/src/test/java/org/apache/beam/it/common/dataflow/AbstractPipelineLauncherTest.java
@@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.it.gcp.dataflow;
+package org.apache.beam.it.common.dataflow;
import static com.google.common.truth.Truth.assertThat;
import static org.junit.Assert.assertThrows;
@@ -34,7 +34,6 @@ import dev.failsafe.FailsafeException;
import java.io.IOException;
import java.net.SocketTimeoutException;
import org.apache.beam.it.common.PipelineLauncher.JobState;
-import org.apache.beam.it.common.dataflow.AbstractPipelineLauncher;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
diff --git
a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/DefaultPipelineLauncherTest.java
b/it/common/src/test/java/org/apache/beam/it/common/dataflow/DefaultPipelineLauncherTest.java
similarity index 92%
rename from
it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/DefaultPipelineLauncherTest.java
rename to
it/common/src/test/java/org/apache/beam/it/common/dataflow/DefaultPipelineLauncherTest.java
index 52ff5c0a230..e9dc6a1a739 100644
---
a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/DefaultPipelineLauncherTest.java
+++
b/it/common/src/test/java/org/apache/beam/it/common/dataflow/DefaultPipelineLauncherTest.java
@@ -15,17 +15,17 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.it.gcp.dataflow;
+package org.apache.beam.it.common.dataflow;
+import static
org.apache.beam.it.common.dataflow.DefaultPipelineLauncher.BEAM_METRICS_NAMESPACE;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.io.IOException;
import java.time.Instant;
+import
org.apache.beam.it.common.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
import org.apache.beam.it.common.PipelineLauncher;
-import org.apache.beam.it.common.dataflow.DefaultPipelineLauncher;
-import org.apache.beam.it.gcp.IOLoadTestBase;
-import org.apache.beam.it.gcp.IOLoadTestBase.PipelineMetricsType;
+import org.apache.beam.it.common.IOLoadTestBase;
import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.testutils.metrics.TimeMonitor;
@@ -49,7 +49,7 @@ public class DefaultPipelineLauncherTest {
pipeline
.apply(GenerateSequence.from(0).to(numElements))
- .apply(ParDo.of(new
TimeMonitor<>(IOLoadTestBase.BEAM_METRICS_NAMESPACE, timeMetrics)))
+ .apply(ParDo.of(new TimeMonitor<>(BEAM_METRICS_NAMESPACE,
timeMetrics)))
.apply(ParDo.of(new IOLoadTestBase.CountingFn<>(counterMetrics)));
PipelineLauncher.LaunchConfig options =
