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
commit 875cfd11cedd5c7b192fbdb98eeb2484e6c01123 Author: Vitaly Terentyev <[email protected]> AuthorDate: Tue Jul 7 15:58:46 2026 +0400 Refactor DefaultPipelineLauncher --- .../common/dataflow/DefaultPipelineLauncher.java | 45 ++++++++++++++-------- .../bigquery/BigQueryResourceManagerTest.java | 4 +- .../bigquery/BigQueryResourceManagerUtilsTest.java | 3 +- .../dataflow/AbstractPipelineLauncherTest.java | 3 +- .../dataflow/DefaultPipelineLauncherTest.java | 9 ++--- .../it/common/{ => dataflow}/IOLoadTestBase.java | 23 +++-------- .../it/common/{ => dataflow}/IOStressTestBase.java | 2 +- .../it/common/{ => dataflow}/LoadTestBase.java | 6 ++- .../beam/it/gcp/datagenerator/package-info.java | 20 ---------- .../java/org/apache/beam/it/gcp/WordCountIT.java | 2 + .../apache/beam/it/gcp/bigquery/BigQueryIOLT.java | 2 +- .../apache/beam/it/gcp/bigquery/BigQueryIOST.java | 2 +- .../beam/it/gcp/bigquery/BigQueryStreamingLT.java | 2 +- .../apache/beam/it/gcp/bigtable/BigTableIOLT.java | 2 +- .../apache/beam/it/gcp/bigtable/BigTableIOST.java | 2 +- .../beam/it/gcp/datagenerator/DataGenerator.java | 2 +- .../org/apache/beam/it/gcp/pubsub/PubSubIOST.java | 2 +- .../org/apache/beam/it/gcp/pubsub/PubsubIOLT.java | 2 +- .../apache/beam/it/gcp/spanner/SpannerIOLT.java | 2 +- .../apache/beam/it/gcp/spanner/SpannerIOST.java | 2 +- .../apache/beam/it/gcp/storage/FileBasedIOLT.java | 2 +- .../org/apache/beam/it/iceberg/IcebergIOLT.java | 2 +- it/kafka/build.gradle | 3 +- .../java/org/apache/beam/it/kafka/KafkaIOLT.java | 3 +- .../java/org/apache/beam/it/kafka/KafkaIOST.java | 3 +- 25 files changed, 65 insertions(+), 85 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/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 93% 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..333549c621b 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,16 @@ * 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.sdk.io.GenerateSequence; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testutils.metrics.TimeMonitor; @@ -49,7 +48,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 = 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/dataflow/IOLoadTestBase.java similarity index 88% rename from it/common/src/test/java/org/apache/beam/it/common/IOLoadTestBase.java rename to it/common/src/test/java/org/apache/beam/it/common/dataflow/IOLoadTestBase.java index fb9feb6663b..8a1a94df1ae 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/dataflow/IOLoadTestBase.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.it.common; +package org.apache.beam.it.common.dataflow; import com.google.cloud.Timestamp; import java.io.IOException; @@ -24,7 +24,9 @@ import java.util.Collection; import java.util.Map; import java.util.UUID; -import org.apache.beam.it.common.dataflow.DefaultPipelineLauncher; +import org.apache.beam.it.common.PipelineLauncher; +import org.apache.beam.it.common.TestProperties; +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 +40,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 +93,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/IOStressTestBase.java b/it/common/src/test/java/org/apache/beam/it/common/dataflow/IOStressTestBase.java similarity index 99% rename from it/common/src/test/java/org/apache/beam/it/common/IOStressTestBase.java rename to it/common/src/test/java/org/apache/beam/it/common/dataflow/IOStressTestBase.java index a15e98d65d8..e08eec031eb 100644 --- a/it/common/src/test/java/org/apache/beam/it/common/IOStressTestBase.java +++ b/it/common/src/test/java/org/apache/beam/it/common/dataflow/IOStressTestBase.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.it.common; +package org.apache.beam.it.common.dataflow; import java.io.Serializable; import java.time.Duration; 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/dataflow/LoadTestBase.java similarity index 99% rename from it/common/src/test/java/org/apache/beam/it/common/LoadTestBase.java rename to it/common/src/test/java/org/apache/beam/it/common/dataflow/LoadTestBase.java index 7d7adbbae0a..36071283506 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/dataflow/LoadTestBase.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.it.common; +package org.apache.beam.it.common.dataflow; import static org.apache.beam.it.common.logging.LogStrings.formatForLogging; import static org.apache.beam.it.common.dataflow.AbstractPipelineLauncher.RUNNER_V2; @@ -41,9 +41,11 @@ import java.util.Map; import java.util.Map.Entry; import java.util.regex.Pattern; +import org.apache.beam.it.common.PipelineLauncher; import org.apache.beam.it.common.PipelineLauncher.LaunchInfo; +import org.apache.beam.it.common.PipelineOperator; +import org.apache.beam.it.common.TestProperties; 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/main/java/org/apache/beam/it/gcp/datagenerator/package-info.java b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datagenerator/package-info.java deleted file mode 100644 index d563025ebeb..00000000000 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datagenerator/package-info.java +++ /dev/null @@ -1,20 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -/** Data generator for load tests. */ -package org.apache.beam.it.gcp.datagenerator; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/WordCountIT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/WordCountIT.java index b561c0c71fb..9719caa59f1 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/WordCountIT.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/WordCountIT.java @@ -25,6 +25,8 @@ import java.io.IOException; import java.time.Duration; import java.util.Arrays; import java.util.List; + +import org.apache.beam.it.common.dataflow.IOLoadTestBase; import org.apache.beam.it.common.PipelineLauncher.LaunchConfig; import org.apache.beam.it.common.PipelineLauncher.LaunchInfo; import org.apache.beam.it.common.PipelineLauncher.Sdk; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryIOLT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryIOLT.java index 9a4af4082e2..b734f93478e 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryIOLT.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryIOLT.java @@ -45,7 +45,7 @@ import org.apache.beam.it.common.PipelineOperator; import org.apache.beam.it.common.TestProperties; import org.apache.beam.it.common.bigquery.BigQueryResourceManager; import org.apache.beam.it.common.utils.ResourceManagerUtils; -import org.apache.beam.it.common.IOLoadTestBase; +import org.apache.beam.it.common.dataflow.IOLoadTestBase; import org.apache.beam.sdk.io.Read; import org.apache.beam.sdk.io.gcp.bigquery.AvroWriteRequest; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryIOST.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryIOST.java index bbe4df61191..4c86eb139bf 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryIOST.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryIOST.java @@ -43,7 +43,7 @@ import org.apache.beam.it.common.PipelineOperator; import org.apache.beam.it.common.TestProperties; import org.apache.beam.it.common.bigquery.BigQueryResourceManager; import org.apache.beam.it.common.utils.ResourceManagerUtils; -import org.apache.beam.it.common.IOStressTestBase; +import org.apache.beam.it.common.dataflow.IOStressTestBase; import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions; import org.apache.beam.sdk.io.Read; import org.apache.beam.sdk.io.gcp.bigquery.AvroWriteRequest; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryStreamingLT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryStreamingLT.java index 7bf759b42f9..75a7be50a50 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryStreamingLT.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryStreamingLT.java @@ -41,7 +41,7 @@ import java.util.stream.Collectors; import org.apache.beam.it.common.PipelineLauncher; import org.apache.beam.it.common.PipelineOperator; import org.apache.beam.it.common.TestProperties; -import org.apache.beam.it.common.IOLoadTestBase; +import org.apache.beam.it.common.dataflow.IOLoadTestBase; import org.apache.beam.runners.dataflow.DataflowRunner; import org.apache.beam.sdk.io.GenerateSequence; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigtable/BigTableIOLT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigtable/BigTableIOLT.java index aea26f2f0a4..03efb42dafb 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigtable/BigTableIOLT.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigtable/BigTableIOLT.java @@ -35,7 +35,7 @@ import org.apache.beam.it.common.PipelineLauncher; import org.apache.beam.it.common.PipelineOperator; import org.apache.beam.it.common.TestProperties; import org.apache.beam.it.common.utils.ResourceManagerUtils; -import org.apache.beam.it.common.IOLoadTestBase; +import org.apache.beam.it.common.dataflow.IOLoadTestBase; import org.apache.beam.sdk.io.GenerateSequence; import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO; import org.apache.beam.sdk.testing.TestPipeline; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigtable/BigTableIOST.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigtable/BigTableIOST.java index 85950772bce..5e70a0e2d58 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigtable/BigTableIOST.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigtable/BigTableIOST.java @@ -38,7 +38,7 @@ import org.apache.beam.it.common.PipelineLauncher; import org.apache.beam.it.common.PipelineOperator; import org.apache.beam.it.common.TestProperties; import org.apache.beam.it.common.utils.ResourceManagerUtils; -import org.apache.beam.it.common.IOStressTestBase; +import org.apache.beam.it.common.dataflow.IOStressTestBase; import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions; import org.apache.beam.sdk.io.Read; import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO; diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datagenerator/DataGenerator.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/datagenerator/DataGenerator.java similarity index 99% rename from it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datagenerator/DataGenerator.java rename to it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/datagenerator/DataGenerator.java index 99016b5dd3a..0b838bddeb6 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datagenerator/DataGenerator.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/datagenerator/DataGenerator.java @@ -17,7 +17,7 @@ */ package org.apache.beam.it.gcp.datagenerator; -import static org.apache.beam.it.gcp.LoadTestBase.createConfig; +import static org.apache.beam.it.common.dataflow.LoadTestBase.createConfig; import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatPipeline; import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatResult; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/pubsub/PubSubIOST.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/pubsub/PubSubIOST.java index 74bd994055a..98092152ad2 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/pubsub/PubSubIOST.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/pubsub/PubSubIOST.java @@ -38,7 +38,7 @@ import org.apache.beam.it.common.PipelineLauncher; import org.apache.beam.it.common.PipelineOperator; import org.apache.beam.it.common.TestProperties; import org.apache.beam.it.common.utils.ResourceManagerUtils; -import org.apache.beam.it.common.IOStressTestBase; +import org.apache.beam.it.common.dataflow.IOStressTestBase; import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions; import org.apache.beam.sdk.extensions.protobuf.Proto3SchemaMessages.Primitive; import org.apache.beam.sdk.io.Read; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/pubsub/PubsubIOLT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/pubsub/PubsubIOLT.java index a9c7b63e64c..360073ac0e9 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/pubsub/PubsubIOLT.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/pubsub/PubsubIOLT.java @@ -38,7 +38,7 @@ import org.apache.beam.it.common.PipelineLauncher; import org.apache.beam.it.common.PipelineOperator; import org.apache.beam.it.common.TestProperties; import org.apache.beam.it.common.utils.ResourceManagerUtils; -import org.apache.beam.it.common.IOLoadTestBase; +import org.apache.beam.it.common.dataflow.IOLoadTestBase; import org.apache.beam.runners.direct.DirectOptions; import org.apache.beam.sdk.extensions.protobuf.Proto3SchemaMessages.Primitive; import org.apache.beam.sdk.io.Read; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/spanner/SpannerIOLT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/spanner/SpannerIOLT.java index a29cf32eae1..99682746d2b 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/spanner/SpannerIOLT.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/spanner/SpannerIOLT.java @@ -38,7 +38,7 @@ import org.apache.beam.it.common.PipelineLauncher; import org.apache.beam.it.common.PipelineOperator; import org.apache.beam.it.common.TestProperties; import org.apache.beam.it.common.utils.ResourceManagerUtils; -import org.apache.beam.it.common.IOLoadTestBase; +import org.apache.beam.it.common.dataflow.IOLoadTestBase; import org.apache.beam.sdk.io.GenerateSequence; import org.apache.beam.sdk.io.gcp.spanner.SpannerIO; import org.apache.beam.sdk.io.synthetic.SyntheticSourceOptions; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/spanner/SpannerIOST.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/spanner/SpannerIOST.java index 8ce34d06de8..01d8deef177 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/spanner/SpannerIOST.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/spanner/SpannerIOST.java @@ -38,7 +38,7 @@ import org.apache.beam.it.common.PipelineLauncher; import org.apache.beam.it.common.PipelineOperator; import org.apache.beam.it.common.TestProperties; import org.apache.beam.it.common.utils.ResourceManagerUtils; -import org.apache.beam.it.common.IOStressTestBase; +import org.apache.beam.it.common.dataflow.IOStressTestBase; import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions; import org.apache.beam.sdk.io.Read; import org.apache.beam.sdk.io.gcp.spanner.SpannerIO; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/FileBasedIOLT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/FileBasedIOLT.java index febcfc01f3f..abec875981f 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/FileBasedIOLT.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/FileBasedIOLT.java @@ -34,7 +34,7 @@ import org.apache.beam.it.common.PipelineLauncher; import org.apache.beam.it.common.PipelineOperator; import org.apache.beam.it.common.TestProperties; import org.apache.beam.it.common.utils.ResourceManagerUtils; -import org.apache.beam.it.common.IOLoadTestBase; +import org.apache.beam.it.common.dataflow.IOLoadTestBase; import org.apache.beam.sdk.io.Compression; import org.apache.beam.sdk.io.Read; import org.apache.beam.sdk.io.TextIO; diff --git a/it/iceberg/src/test/java/org/apache/beam/it/iceberg/IcebergIOLT.java b/it/iceberg/src/test/java/org/apache/beam/it/iceberg/IcebergIOLT.java index d2e342ccecb..3f486ec3f5d 100644 --- a/it/iceberg/src/test/java/org/apache/beam/it/iceberg/IcebergIOLT.java +++ b/it/iceberg/src/test/java/org/apache/beam/it/iceberg/IcebergIOLT.java @@ -1,6 +1,6 @@ package org.apache.beam.it.iceberg; -import org.apache.beam.it.common.IOLoadTestBase; +import org.apache.beam.it.common.dataflow.IOLoadTestBase; public class IcebergIOLT extends IOLoadTestBase { diff --git a/it/kafka/build.gradle b/it/kafka/build.gradle index 0496c423beb..32ef0623fea 100644 --- a/it/kafka/build.gradle +++ b/it/kafka/build.gradle @@ -42,8 +42,7 @@ dependencies { permitUsedUndeclared library.java.guava testImplementation library.java.mockito_core testRuntimeOnly library.java.slf4j_simple - // TODO: Remove the below dependencies after KafkaIOLT has been moved to the right directory. - testImplementation project(path: ":it:google-cloud-platform") + testImplementation project(path: ":it:common", configuration: "testRuntimeMigration") testImplementation project(path: ":sdks:java:io:kafka") testImplementation project(path: ":sdks:java:io:synthetic") testImplementation project(path: ":runners:direct-java") diff --git a/it/kafka/src/test/java/org/apache/beam/it/kafka/KafkaIOLT.java b/it/kafka/src/test/java/org/apache/beam/it/kafka/KafkaIOLT.java index c20699d61a5..4b0ecbfd343 100644 --- a/it/kafka/src/test/java/org/apache/beam/it/kafka/KafkaIOLT.java +++ b/it/kafka/src/test/java/org/apache/beam/it/kafka/KafkaIOLT.java @@ -27,11 +27,12 @@ import java.time.ZoneOffset; import java.time.format.DateTimeFormatter; import java.util.Map; import java.util.UUID; +import org.apache.beam.it.common.dataflow.DefaultPipelineLauncher.PipelineMetricsType; import org.apache.beam.it.common.PipelineLauncher; import org.apache.beam.it.common.PipelineOperator; import org.apache.beam.it.common.TestProperties; import org.apache.beam.it.common.utils.ResourceManagerUtils; -import org.apache.beam.it.gcp.IOLoadTestBase; +import org.apache.beam.it.common.dataflow.IOLoadTestBase; import org.apache.beam.runners.direct.DirectOptions; import org.apache.beam.sdk.io.kafka.KafkaIO; import org.apache.beam.sdk.io.synthetic.SyntheticOptions; diff --git a/it/kafka/src/test/java/org/apache/beam/it/kafka/KafkaIOST.java b/it/kafka/src/test/java/org/apache/beam/it/kafka/KafkaIOST.java index bb1cb37a11b..cfc11127c34 100644 --- a/it/kafka/src/test/java/org/apache/beam/it/kafka/KafkaIOST.java +++ b/it/kafka/src/test/java/org/apache/beam/it/kafka/KafkaIOST.java @@ -31,10 +31,11 @@ import java.util.List; import java.util.Map; import java.util.Objects; import java.util.UUID; +import org.apache.beam.it.common.dataflow.DefaultPipelineLauncher.PipelineMetricsType; import org.apache.beam.it.common.PipelineLauncher; import org.apache.beam.it.common.PipelineOperator; import org.apache.beam.it.common.TestProperties; -import org.apache.beam.it.gcp.IOStressTestBase; +import org.apache.beam.it.common.dataflow.IOStressTestBase; import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions; import org.apache.beam.sdk.io.Read; import org.apache.beam.sdk.io.kafka.KafkaIO;
