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 02b61154238 Add IcebergIOLT
02b61154238 is described below
commit 02b61154238fcd559b07104eea224ee99ff7ef1e
Author: Vitaly Terentyev <[email protected]>
AuthorDate: Tue Jul 7 17:04:31 2026 +0400
Add IcebergIOLT
---
it/common/build.gradle | 3 +-
.../apache/beam/it/common}/artifacts/Artifact.java | 2 +-
.../beam/it/common}/artifacts/ArtifactClient.java | 2 +-
.../beam/it/common}/artifacts/GcsArtifact.java | 2 +-
.../artifacts/matchers/ArtifactAsserts.java | 13 +-
.../artifacts/matchers/ArtifactsSubject.java | 15 +-
.../common}/artifacts/matchers/package-info.java | 2 +-
.../beam/it/common}/artifacts/package-info.java | 2 +-
.../it/common}/artifacts/utils/ArtifactUtils.java | 2 +-
.../it/common}/artifacts/utils/AvroTestUtil.java | 2 +-
.../it/common}/artifacts/utils/JsonTestUtil.java | 2 +-
.../common}/artifacts/utils/ParquetTestUtil.java | 2 +-
.../it/common}/artifacts/utils/package-info.java | 2 +-
.../common/dataflow/DefaultPipelineLauncher.java | 23 +-
.../it/common}/storage/GcsResourceManager.java | 10 +-
.../beam/it/common}/storage/package-info.java | 2 +-
.../beam/it/common}/artifacts/GcsArtifactTest.java | 4 +-
.../common}/artifacts/utils/ArtifactUtilsTest.java | 4 +-
.../bigquery/BigQueryResourceManagerUtilsTest.java | 1 -
.../dataflow/DefaultPipelineLauncherTest.java | 2 +-
.../beam/it/common/dataflow/IOLoadTestBase.java | 5 +-
.../beam/it/common/dataflow/LoadTestBase.java | 3 +-
.../it/common}/storage/GcsResourceManagerTest.java | 19 +-
.../gcp/bigquery/conditions/BigQueryRowsCheck.java | 2 +-
.../it/gcp/dataflow/ClassicTemplateClient.java | 1 -
.../beam/it/gcp/dataflow/FlexTemplateClient.java | 1 -
.../it/gcp/spanner/matchers/SpannerAsserts.java | 2 +-
.../java/org/apache/beam/it/gcp/WordCountIT.java | 3 +-
.../apache/beam/it/gcp/bigquery/BigQueryIOLT.java | 3 +-
.../apache/beam/it/gcp/bigquery/BigQueryIOST.java | 3 +-
.../apache/beam/it/gcp/bigtable/BigTableIOLT.java | 3 +-
.../apache/beam/it/gcp/bigtable/BigTableIOST.java | 3 +-
.../org/apache/beam/it/gcp/pubsub/PubSubIOST.java | 3 +-
.../org/apache/beam/it/gcp/pubsub/PubsubIOLT.java | 3 +-
.../apache/beam/it/gcp/spanner/SpannerIOLT.java | 3 +-
.../apache/beam/it/gcp/spanner/SpannerIOST.java | 3 +-
.../apache/beam/it/gcp/storage/FileBasedIOLT.java | 4 +-
it/iceberg/build.gradle | 1 +
.../org/apache/beam/it/iceberg/IcebergIOLT.java | 353 ++++++++++++++++++++-
.../java/org/apache/beam/it/kafka/KafkaIOLT.java | 4 +-
.../java/org/apache/beam/it/kafka/KafkaIOST.java | 2 +-
41 files changed, 436 insertions(+), 85 deletions(-)
diff --git a/it/common/build.gradle b/it/common/build.gradle
index 7b8559aa6e1..dc789c08fd3 100644
--- a/it/common/build.gradle
+++ b/it/common/build.gradle
@@ -32,6 +32,7 @@ dependencies {
implementation project(path: ":sdks:java:testing:test-utils")
implementation 'com.google.cloud:google-cloud-bigquery'
implementation 'com.google.cloud:google-cloud-monitoring'
+ implementation 'com.google.cloud:google-cloud-storage'
provided 'com.google.api.grpc:proto-google-cloud-monitoring-v3'
implementation library.java.gax
implementation library.java.google_api_services_dataflow
@@ -47,12 +48,12 @@ dependencies {
implementation library.java.guava
implementation library.java.protobuf_java_util
implementation library.java.protobuf_java
+ implementation library.java.junit
// TODO: excluding Guava until Truth updates it to >32.1.x
testImplementation(library.java.truth) {
exclude group: 'com.google.guava', module: 'guava'
}
- testImplementation library.java.junit
testImplementation library.java.mockito_inline
testRuntimeOnly library.java.slf4j_simple
testImplementation project(path: ":sdks:java:extensions:protobuf",
configuration: "testRuntimeMigration")
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/Artifact.java
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/Artifact.java
similarity index 97%
rename from
it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/Artifact.java
rename to
it/common/src/main/java/org/apache/beam/it/common/artifacts/Artifact.java
index 830f44598be..6371d6eedb1 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/Artifact.java
+++ b/it/common/src/main/java/org/apache/beam/it/common/artifacts/Artifact.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.artifacts;
+package org.apache.beam.it.common.artifacts;
/**
* Represents a single artifact.
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/ArtifactClient.java
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/ArtifactClient.java
similarity index 99%
rename from
it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/ArtifactClient.java
rename to
it/common/src/main/java/org/apache/beam/it/common/artifacts/ArtifactClient.java
index e463baffeaf..b275d35703e 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/ArtifactClient.java
+++
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/ArtifactClient.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.artifacts;
+package org.apache.beam.it.common.artifacts;
import java.io.IOException;
import java.nio.file.Path;
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/GcsArtifact.java
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/GcsArtifact.java
similarity index 96%
rename from
it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/GcsArtifact.java
rename to
it/common/src/main/java/org/apache/beam/it/common/artifacts/GcsArtifact.java
index c60e3a0fe53..5f4360ddbb2 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/GcsArtifact.java
+++
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/GcsArtifact.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.artifacts;
+package org.apache.beam.it.common.artifacts;
import com.google.cloud.storage.Blob;
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/matchers/ArtifactAsserts.java
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/matchers/ArtifactAsserts.java
similarity index 81%
rename from
it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/matchers/ArtifactAsserts.java
rename to
it/common/src/main/java/org/apache/beam/it/common/artifacts/matchers/ArtifactAsserts.java
index 3b1251307d0..48d04edadc7 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/matchers/ArtifactAsserts.java
+++
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/matchers/ArtifactAsserts.java
@@ -15,14 +15,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.it.gcp.artifacts.matchers;
-
-import static com.google.common.truth.Truth.assertAbout;
-import static
org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatRecords;
+package org.apache.beam.it.common.artifacts.matchers;
import java.util.List;
import org.apache.avro.generic.GenericRecord;
-import org.apache.beam.it.gcp.artifacts.Artifact;
+import org.apache.beam.it.common.artifacts.Artifact;
import org.apache.beam.it.truthmatchers.RecordsSubject;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
@@ -36,7 +33,7 @@ public class ArtifactAsserts {
* @return Truth Subject to chain assertions.
*/
public static ArtifactsSubject assertThatArtifacts(List<Artifact> artifacts)
{
- return assertAbout(ArtifactsSubject.records()).that(artifacts);
+ return Truth.assertAbout(ArtifactsSubject.records()).that(artifacts);
}
/**
@@ -47,7 +44,7 @@ public class ArtifactAsserts {
* @return Truth Subject to chain assertions.
*/
public static ArtifactsSubject assertThatArtifact(Artifact artifact) {
- return
assertAbout(ArtifactsSubject.records()).that(ImmutableList.of(artifact));
+ return
Truth.assertAbout(ArtifactsSubject.records()).that(ImmutableList.of(artifact));
}
/**
@@ -57,6 +54,6 @@ public class ArtifactAsserts {
* @return Truth Subject to chain assertions.
*/
public static RecordsSubject assertThatGenericRecords(List<GenericRecord>
records) {
- return assertThatRecords(ArtifactsSubject.genericRecordToRecords(records));
+ return
PipelineAsserts.assertThatRecords(ArtifactsSubject.genericRecordToRecords(records));
}
}
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/matchers/ArtifactsSubject.java
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/matchers/ArtifactsSubject.java
similarity index 91%
rename from
it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/matchers/ArtifactsSubject.java
rename to
it/common/src/main/java/org/apache/beam/it/common/artifacts/matchers/ArtifactsSubject.java
index ab5b4969948..7a49bd22ff9 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/matchers/ArtifactsSubject.java
+++
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/matchers/ArtifactsSubject.java
@@ -15,10 +15,9 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.it.gcp.artifacts.matchers;
+package org.apache.beam.it.common.artifacts.matchers;
-import static
org.apache.beam.it.gcp.artifacts.matchers.ArtifactAsserts.assertThatGenericRecords;
-import static
org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatRecords;
+import static
org.apache.beam.it.common.artifacts.matchers.ArtifactAsserts.assertThatGenericRecords;
import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.hash.Hashing.sha256;
import com.fasterxml.jackson.core.type.TypeReference;
@@ -31,10 +30,10 @@ import java.util.List;
import java.util.Map;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
-import org.apache.beam.it.gcp.artifacts.Artifact;
-import org.apache.beam.it.gcp.artifacts.utils.AvroTestUtil;
-import org.apache.beam.it.gcp.artifacts.utils.JsonTestUtil;
-import org.apache.beam.it.gcp.artifacts.utils.ParquetTestUtil;
+import org.apache.beam.it.common.artifacts.Artifact;
+import org.apache.beam.it.common.artifacts.utils.AvroTestUtil;
+import org.apache.beam.it.common.artifacts.utils.JsonTestUtil;
+import org.apache.beam.it.common.artifacts.utils.ParquetTestUtil;
import org.apache.beam.it.truthmatchers.RecordsSubject;
/**
@@ -167,6 +166,6 @@ public final class ArtifactsSubject extends Subject {
throw new RuntimeException("Error reading " + artifact.name() + " as
JSON.", e);
}
}
- return assertThatRecords(allRecords);
+ return PipelineAsserts.assertThatRecords(allRecords);
}
}
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/matchers/package-info.java
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/matchers/package-info.java
similarity index 94%
rename from
it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/matchers/package-info.java
rename to
it/common/src/main/java/org/apache/beam/it/common/artifacts/matchers/package-info.java
index 9759071b397..1167e1b6cc2 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/matchers/package-info.java
+++
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/matchers/package-info.java
@@ -17,4 +17,4 @@
*/
/** Package for Artifact Truth matchers / subjects to have reusable
assertions. */
-package org.apache.beam.it.gcp.artifacts.matchers;
+package org.apache.beam.it.common.artifacts.matchers;
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/package-info.java
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/package-info.java
similarity index 95%
rename from
it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/package-info.java
rename to
it/common/src/main/java/org/apache/beam/it/common/artifacts/package-info.java
index de2a2541d55..44b54f82aeb 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/package-info.java
+++
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/package-info.java
@@ -17,4 +17,4 @@
*/
/** Package for working with test artifacts. */
-package org.apache.beam.it.gcp.artifacts;
+package org.apache.beam.it.common.artifacts;
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/utils/ArtifactUtils.java
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/utils/ArtifactUtils.java
similarity index 98%
rename from
it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/utils/ArtifactUtils.java
rename to
it/common/src/main/java/org/apache/beam/it/common/artifacts/utils/ArtifactUtils.java
index 854651ae1f1..3a3bcb2d7ec 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/utils/ArtifactUtils.java
+++
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/utils/ArtifactUtils.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.artifacts.utils;
+package org.apache.beam.it.common.artifacts.utils;
import static java.util.Arrays.stream;
import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/utils/AvroTestUtil.java
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/utils/AvroTestUtil.java
similarity index 98%
rename from
it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/utils/AvroTestUtil.java
rename to
it/common/src/main/java/org/apache/beam/it/common/artifacts/utils/AvroTestUtil.java
index 5cfe4c137e3..58241f9a1db 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/utils/AvroTestUtil.java
+++
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/utils/AvroTestUtil.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.artifacts.utils;
+package org.apache.beam.it.common.artifacts.utils;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/utils/JsonTestUtil.java
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/utils/JsonTestUtil.java
similarity index 99%
rename from
it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/utils/JsonTestUtil.java
rename to
it/common/src/main/java/org/apache/beam/it/common/artifacts/utils/JsonTestUtil.java
index 9a83558f7bf..08c3cdfb0c5 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/utils/JsonTestUtil.java
+++
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/utils/JsonTestUtil.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.artifacts.utils;
+package org.apache.beam.it.common.artifacts.utils;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/utils/ParquetTestUtil.java
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/utils/ParquetTestUtil.java
similarity index 98%
rename from
it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/utils/ParquetTestUtil.java
rename to
it/common/src/main/java/org/apache/beam/it/common/artifacts/utils/ParquetTestUtil.java
index d6ebf825e2c..fb4326e1b4a 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/utils/ParquetTestUtil.java
+++
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/utils/ParquetTestUtil.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.artifacts.utils;
+package org.apache.beam.it.common.artifacts.utils;
import java.io.ByteArrayInputStream;
import java.io.File;
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/utils/package-info.java
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/utils/package-info.java
similarity index 94%
rename from
it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/utils/package-info.java
rename to
it/common/src/main/java/org/apache/beam/it/common/artifacts/utils/package-info.java
index 53cb4ed1d3a..aaf820b2510 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/utils/package-info.java
+++
b/it/common/src/main/java/org/apache/beam/it/common/artifacts/utils/package-info.java
@@ -17,4 +17,4 @@
*/
/** Package for artifact utilities. */
-package org.apache.beam.it.gcp.artifacts.utils;
+package org.apache.beam.it.common.artifacts.utils;
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 32a4eb6eb1b..af6438a561a 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
@@ -192,15 +192,13 @@ public class DefaultPipelineLauncher extends
AbstractPipelineLauncher {
.metrics()
.queryMetrics(
MetricsFilter.builder()
- .addNameFilter(
- MetricNameFilter.named(BEAM_METRICS_NAMESPACE,
metricName))
+
.addNameFilter(MetricNameFilter.named(BEAM_METRICS_NAMESPACE, metricName))
.build());
return metrics.getDistributions();
}
/** Pull Beam pipeline defined metrics given the jobId. */
- public Long getBeamMetric(
- String jobId, PipelineMetricsType metricType, String metricName) {
+ public Long getBeamMetric(String jobId, PipelineMetricsType metricType,
String metricName) {
PipelineResult pipelineResult =
MANAGED_JOBS.getOrDefault(jobId, UNMANAGED_JOBS.getOrDefault(jobId,
null));
if (pipelineResult != null) {
@@ -209,8 +207,7 @@ public class DefaultPipelineLauncher extends
AbstractPipelineLauncher {
.metrics()
.queryMetrics(
MetricsFilter.builder()
- .addNameFilter(
- MetricNameFilter.named(BEAM_METRICS_NAMESPACE,
metricName))
+
.addNameFilter(MetricNameFilter.named(BEAM_METRICS_NAMESPACE, metricName))
.build());
switch (metricType) {
@@ -222,9 +219,7 @@ public class DefaultPipelineLauncher extends
AbstractPipelineLauncher {
return metricResult.getAttempted();
} catch (NoSuchElementException e) {
LOG.error(
- "Failed to get metric {}, from namespace {}",
- metricName,
- BEAM_METRICS_NAMESPACE);
+ "Failed to get metric {}, from namespace {}", metricName,
BEAM_METRICS_NAMESPACE);
}
return UNKNOWN_METRIC_VALUE;
case STARTTIME:
@@ -273,8 +268,7 @@ public class DefaultPipelineLauncher extends
AbstractPipelineLauncher {
String.format(
"Invalid Beam metrics name: %s, expected:
'%s:metric_type:metric_name'",
metricName, BEAM_METRICS_NAMESPACE));
- PipelineMetricsType metricType =
- PipelineMetricsType.valueOf(nameSpacedMetrics[1]);
+ PipelineMetricsType metricType =
PipelineMetricsType.valueOf(nameSpacedMetrics[1]);
// Pipeline defined metrics are long values. Have to cast to double that
is what the base
// class defined.
@@ -449,7 +443,9 @@ 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(org.apache.beam.sdk.testing.TestPipeline.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));
@@ -457,7 +453,8 @@ public class DefaultPipelineLauncher extends
AbstractPipelineLauncher {
throw new RuntimeException(
"Unable to instantiate test options from system property "
+
org.apache.beam.sdk.testing.TestPipeline.PROPERTY_BEAM_TEST_PIPELINE_OPTIONS
- + ":" + beamTestPipelineOptions,
+ + ":"
+ + beamTestPipelineOptions,
e);
}
}
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/storage/GcsResourceManager.java
b/it/common/src/main/java/org/apache/beam/it/common/storage/GcsResourceManager.java
similarity index 97%
rename from
it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/storage/GcsResourceManager.java
rename to
it/common/src/main/java/org/apache/beam/it/common/storage/GcsResourceManager.java
index 730ca23ee54..85f2c8b0b01 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/storage/GcsResourceManager.java
+++
b/it/common/src/main/java/org/apache/beam/it/common/storage/GcsResourceManager.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.storage;
+package org.apache.beam.it.common.storage;
import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
@@ -39,10 +39,10 @@ import java.util.List;
import java.util.function.Consumer;
import java.util.regex.Pattern;
import org.apache.beam.it.common.ResourceManager;
-import org.apache.beam.it.gcp.artifacts.Artifact;
-import org.apache.beam.it.gcp.artifacts.ArtifactClient;
-import org.apache.beam.it.gcp.artifacts.GcsArtifact;
-import org.apache.beam.it.gcp.artifacts.utils.ArtifactUtils;
+import org.apache.beam.it.common.artifacts.Artifact;
+import org.apache.beam.it.common.artifacts.ArtifactClient;
+import org.apache.beam.it.common.artifacts.GcsArtifact;
+import org.apache.beam.it.common.artifacts.utils.ArtifactUtils;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
import org.junit.rules.TestName;
import org.slf4j.Logger;
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/storage/package-info.java
b/it/common/src/main/java/org/apache/beam/it/common/storage/package-info.java
similarity index 95%
rename from
it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/storage/package-info.java
rename to
it/common/src/main/java/org/apache/beam/it/common/storage/package-info.java
index efcea3aba26..1c5b3b3ce21 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/storage/package-info.java
+++
b/it/common/src/main/java/org/apache/beam/it/common/storage/package-info.java
@@ -17,4 +17,4 @@
*/
/** Package for managing Google Cloud Storage resources within integration
tests. */
-package org.apache.beam.it.gcp.storage;
+package org.apache.beam.it.common.storage;
diff --git
a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/artifacts/GcsArtifactTest.java
b/it/common/src/test/java/org/apache/beam/it/common/artifacts/GcsArtifactTest.java
similarity index 94%
rename from
it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/artifacts/GcsArtifactTest.java
rename to
it/common/src/test/java/org/apache/beam/it/common/artifacts/GcsArtifactTest.java
index 59b7931b603..470741b399a 100644
---
a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/artifacts/GcsArtifactTest.java
+++
b/it/common/src/test/java/org/apache/beam/it/common/artifacts/GcsArtifactTest.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.artifacts;
+package org.apache.beam.it.common.artifacts;
import static com.google.common.truth.Truth.assertThat;
import static org.mockito.Mockito.when;
@@ -30,7 +30,7 @@ import org.mockito.Mock;
import org.mockito.junit.MockitoJUnit;
import org.mockito.junit.MockitoRule;
-/** Unit tests for {@link GcsArtifact}. */
+/** Unit tests for {@link org.apache.beam.it.common.artifacts.GcsArtifact}. */
@RunWith(JUnit4.class)
public class GcsArtifactTest {
@Rule public final MockitoRule mockito = MockitoJUnit.rule();
diff --git
a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/artifacts/utils/ArtifactUtilsTest.java
b/it/common/src/test/java/org/apache/beam/it/common/artifacts/utils/ArtifactUtilsTest.java
similarity index 94%
rename from
it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/artifacts/utils/ArtifactUtilsTest.java
rename to
it/common/src/test/java/org/apache/beam/it/common/artifacts/utils/ArtifactUtilsTest.java
index 491d32d0ce6..137c35aaf4b 100644
---
a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/artifacts/utils/ArtifactUtilsTest.java
+++
b/it/common/src/test/java/org/apache/beam/it/common/artifacts/utils/ArtifactUtilsTest.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.artifacts.utils;
+package org.apache.beam.it.common.artifacts.utils;
import static com.google.common.truth.Truth.assertThat;
import static org.junit.Assert.assertThrows;
@@ -24,7 +24,7 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
-/** Artifacts for {@link ArtifactUtils}. */
+/** Artifacts for {@link
org.apache.beam.it.common.artifacts.utils.ArtifactUtils}. */
@RunWith(JUnit4.class)
public final class ArtifactUtilsTest {
diff --git
a/it/common/src/test/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerUtilsTest.java
b/it/common/src/test/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerUtilsTest.java
index eaff8074072..7fc848b4529 100644
---
a/it/common/src/test/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerUtilsTest.java
+++
b/it/common/src/test/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerUtilsTest.java
@@ -21,7 +21,6 @@ import static
org.apache.beam.it.common.bigquery.BigQueryResourceManagerUtils.ch
import static org.junit.Assert.assertThrows;
import java.util.Arrays;
-
import org.junit.Test;
/** Unit tests for {@link BigQueryResourceManagerUtils}. */
diff --git
a/it/common/src/test/java/org/apache/beam/it/common/dataflow/DefaultPipelineLauncherTest.java
b/it/common/src/test/java/org/apache/beam/it/common/dataflow/DefaultPipelineLauncherTest.java
index 333549c621b..35510a0f1d1 100644
---
a/it/common/src/test/java/org/apache/beam/it/common/dataflow/DefaultPipelineLauncherTest.java
+++
b/it/common/src/test/java/org/apache/beam/it/common/dataflow/DefaultPipelineLauncherTest.java
@@ -23,8 +23,8 @@ 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.PipelineMetricsType;
import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.testutils.metrics.TimeMonitor;
diff --git
a/it/common/src/test/java/org/apache/beam/it/common/dataflow/IOLoadTestBase.java
b/it/common/src/test/java/org/apache/beam/it/common/dataflow/IOLoadTestBase.java
index 8a1a94df1ae..c35f402bd8a 100644
---
a/it/common/src/test/java/org/apache/beam/it/common/dataflow/IOLoadTestBase.java
+++
b/it/common/src/test/java/org/apache/beam/it/common/dataflow/IOLoadTestBase.java
@@ -17,13 +17,14 @@
*/
package org.apache.beam.it.common.dataflow;
+import static
org.apache.beam.it.common.dataflow.DefaultPipelineLauncher.BEAM_METRICS_NAMESPACE;
+
import com.google.cloud.Timestamp;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Map;
import java.util.UUID;
-
import org.apache.beam.it.common.PipelineLauncher;
import org.apache.beam.it.common.TestProperties;
import
org.apache.beam.it.common.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
@@ -40,8 +41,6 @@ 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({
diff --git
a/it/common/src/test/java/org/apache/beam/it/common/dataflow/LoadTestBase.java
b/it/common/src/test/java/org/apache/beam/it/common/dataflow/LoadTestBase.java
index 36071283506..cd1b71dc755 100644
---
a/it/common/src/test/java/org/apache/beam/it/common/dataflow/LoadTestBase.java
+++
b/it/common/src/test/java/org/apache/beam/it/common/dataflow/LoadTestBase.java
@@ -17,8 +17,8 @@
*/
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;
+import static org.apache.beam.it.common.logging.LogStrings.formatForLogging;
import com.google.api.gax.core.CredentialsProvider;
import com.google.api.gax.core.FixedCredentialsProvider;
@@ -40,7 +40,6 @@ import java.util.List;
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;
diff --git
a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/GcsResourceManagerTest.java
b/it/common/src/test/java/org/apache/beam/it/common/storage/GcsResourceManagerTest.java
similarity index 95%
rename from
it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/GcsResourceManagerTest.java
rename to
it/common/src/test/java/org/apache/beam/it/common/storage/GcsResourceManagerTest.java
index 0153573feae..70dc6dcbc9c 100644
---
a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/GcsResourceManagerTest.java
+++
b/it/common/src/test/java/org/apache/beam/it/common/storage/GcsResourceManagerTest.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.storage;
+package org.apache.beam.it.common.storage;
import static com.google.common.truth.Truth.assertThat;
import static org.junit.Assert.assertThrows;
@@ -35,6 +35,7 @@ import com.google.cloud.storage.BlobInfo;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.Storage.BlobListOption;
import com.google.cloud.storage.Storage.BucketListOption;
+import com.google.common.truth.Truth;
import java.io.IOException;
import java.net.URISyntaxException;
import java.nio.file.Files;
@@ -43,8 +44,8 @@ import java.nio.file.Paths;
import java.util.List;
import java.util.UUID;
import java.util.regex.Pattern;
-import org.apache.beam.it.gcp.artifacts.Artifact;
-import org.apache.beam.it.gcp.artifacts.GcsArtifact;
+import org.apache.beam.it.common.artifacts.Artifact;
+import org.apache.beam.it.common.artifacts.GcsArtifact;
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.io.Resources;
import org.junit.After;
@@ -128,7 +129,7 @@ public final class GcsResourceManagerTest {
verify(client).create(blobInfoCaptor.capture(), contentsCaptor.capture());
BlobInfo actualInfo = blobInfoCaptor.getValue();
- assertThat(actual.blob).isSameInstanceAs(blob);
+ Truth.assertThat(actual.blob).isSameInstanceAs(blob);
assertThat(actualInfo.getBucket()).isEqualTo(BUCKET);
assertThat(actualInfo.getName())
.isEqualTo(String.format("%s/%s/%s", TEST_CLASS, gcsClient.runId(),
artifactName));
@@ -144,7 +145,7 @@ public final class GcsResourceManagerTest {
verify(client).create(blobInfoCaptor.capture(), contentsCaptor.capture());
BlobInfo actualInfo = blobInfoCaptor.getValue();
- assertThat(actual.blob).isSameInstanceAs(blob);
+ Truth.assertThat(actual.blob).isSameInstanceAs(blob);
assertThat(actualInfo.getBucket()).isEqualTo(BUCKET);
assertThat(actualInfo.getName())
.isEqualTo(String.format("%s/%s/%s", TEST_CLASS, gcsClient.runId(),
ARTIFACT_NAME));
@@ -185,8 +186,8 @@ public final class GcsResourceManagerTest {
BlobListOption actualOptions = listOptionsCaptor.getValue();
assertThat(actual).hasSize(2);
- assertThat(actual.get(0).name()).isEqualTo(name1);
- assertThat(actual.get(1).name()).isEqualTo(name3);
+ Truth.assertThat(actual.get(0).name()).isEqualTo(name1);
+ Truth.assertThat(actual.get(1).name()).isEqualTo(name3);
assertThat(actualBucket).isEqualTo(BUCKET);
assertThat(actualOptions)
.isEqualTo(
@@ -221,8 +222,8 @@ public final class GcsResourceManagerTest {
BlobListOption actualOptions = listOptionsCaptor.getValue();
assertThat(actual).hasSize(2);
- assertThat(actual.get(0).name()).isEqualTo(name1);
- assertThat(actual.get(1).name()).isEqualTo(name3);
+ Truth.assertThat(actual.get(0).name()).isEqualTo(name1);
+ Truth.assertThat(actual.get(1).name()).isEqualTo(name3);
assertThat(actualBucket).isEqualTo(BUCKET);
assertThat(actualOptions)
.isEqualTo(
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/bigquery/conditions/BigQueryRowsCheck.java
b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/bigquery/conditions/BigQueryRowsCheck.java
index 57b7d5ac99c..da0cf5e55db 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/bigquery/conditions/BigQueryRowsCheck.java
+++
b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/bigquery/conditions/BigQueryRowsCheck.java
@@ -19,8 +19,8 @@ package org.apache.beam.it.gcp.bigquery.conditions;
import com.google.auto.value.AutoValue;
import com.google.cloud.bigquery.TableId;
-import org.apache.beam.it.conditions.ConditionCheck;
import org.apache.beam.it.common.bigquery.BigQueryResourceManager;
+import org.apache.beam.it.conditions.ConditionCheck;
import org.checkerframework.checker.nullness.qual.Nullable;
/** ConditionCheck to validate if BigQuery has received a certain number of
rows. */
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/ClassicTemplateClient.java
b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/ClassicTemplateClient.java
index 296f7092cfa..7908246dce1 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/ClassicTemplateClient.java
+++
b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/ClassicTemplateClient.java
@@ -30,7 +30,6 @@ import com.google.auth.Credentials;
import com.google.auth.http.HttpCredentialsAdapter;
import dev.failsafe.Failsafe;
import java.io.IOException;
-
import org.apache.beam.it.common.dataflow.AbstractPipelineLauncher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/FlexTemplateClient.java
b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/FlexTemplateClient.java
index 3e7270dbd34..496c4b0e980 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/FlexTemplateClient.java
+++
b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/FlexTemplateClient.java
@@ -32,7 +32,6 @@ import com.google.auth.Credentials;
import com.google.auth.http.HttpCredentialsAdapter;
import dev.failsafe.Failsafe;
import java.io.IOException;
-
import org.apache.beam.it.common.dataflow.AbstractPipelineLauncher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/spanner/matchers/SpannerAsserts.java
b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/spanner/matchers/SpannerAsserts.java
index 5a101e08d37..1778d70cf60 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/spanner/matchers/SpannerAsserts.java
+++
b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/spanner/matchers/SpannerAsserts.java
@@ -17,7 +17,7 @@
*/
package org.apache.beam.it.gcp.spanner.matchers;
-import static
org.apache.beam.it.gcp.artifacts.utils.JsonTestUtil.parseJsonString;
+import static
org.apache.beam.it.common.artifacts.utils.JsonTestUtil.parseJsonString;
import static
org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatRecords;
import com.google.cloud.spanner.Mutation;
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 9719caa59f1..16cf2b8f4d3 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,12 +25,11 @@ 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;
import org.apache.beam.it.common.PipelineOperator.Result;
+import org.apache.beam.it.common.dataflow.IOLoadTestBase;
import org.apache.beam.runners.dataflow.DataflowRunner;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptions;
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 b734f93478e..9ac03db1c74 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
@@ -44,8 +44,9 @@ 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.bigquery.BigQueryResourceManager;
-import org.apache.beam.it.common.utils.ResourceManagerUtils;
+import
org.apache.beam.it.common.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
import org.apache.beam.it.common.dataflow.IOLoadTestBase;
+import org.apache.beam.it.common.utils.ResourceManagerUtils;
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 4c86eb139bf..495a4d4485d 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
@@ -42,8 +42,9 @@ 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.bigquery.BigQueryResourceManager;
-import org.apache.beam.it.common.utils.ResourceManagerUtils;
+import
org.apache.beam.it.common.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
import org.apache.beam.it.common.dataflow.IOStressTestBase;
+import org.apache.beam.it.common.utils.ResourceManagerUtils;
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/bigtable/BigTableIOLT.java
b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigtable/BigTableIOLT.java
index 03efb42dafb..27773121e79 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
@@ -34,8 +34,9 @@ import java.util.Random;
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.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
import org.apache.beam.it.common.dataflow.IOLoadTestBase;
+import org.apache.beam.it.common.utils.ResourceManagerUtils;
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 5e70a0e2d58..b86f13a0de1 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
@@ -37,8 +37,9 @@ import java.util.UUID;
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.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
import org.apache.beam.it.common.dataflow.IOStressTestBase;
+import org.apache.beam.it.common.utils.ResourceManagerUtils;
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/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 98092152ad2..47058c75e39 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
@@ -37,8 +37,9 @@ import java.util.Objects;
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.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
import org.apache.beam.it.common.dataflow.IOStressTestBase;
+import org.apache.beam.it.common.utils.ResourceManagerUtils;
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 360073ac0e9..da43fca2eaa 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
@@ -37,8 +37,9 @@ import java.util.Objects;
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.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
import org.apache.beam.it.common.dataflow.IOLoadTestBase;
+import org.apache.beam.it.common.utils.ResourceManagerUtils;
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 99682746d2b..4441b8dcd6b 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
@@ -37,8 +37,9 @@ import java.util.UUID;
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.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
import org.apache.beam.it.common.dataflow.IOLoadTestBase;
+import org.apache.beam.it.common.utils.ResourceManagerUtils;
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 01d8deef177..4eb7030c23c 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
@@ -37,8 +37,9 @@ import java.util.UUID;
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.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
import org.apache.beam.it.common.dataflow.IOStressTestBase;
+import org.apache.beam.it.common.utils.ResourceManagerUtils;
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 abec875981f..ac1a7fc103c 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
@@ -33,8 +33,10 @@ import java.util.UUID;
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.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
import org.apache.beam.it.common.dataflow.IOLoadTestBase;
+import org.apache.beam.it.common.storage.GcsResourceManager;
+import org.apache.beam.it.common.utils.ResourceManagerUtils;
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/build.gradle b/it/iceberg/build.gradle
index 1a61f86367b..0d4e648151b 100644
--- a/it/iceberg/build.gradle
+++ b/it/iceberg/build.gradle
@@ -135,6 +135,7 @@ dependencies {
testImplementation project(path: ":sdks:java:extensions:protobuf",
configuration: "testRuntimeMigration")
testImplementation project(path: ":sdks:java:io:synthetic")
+ testImplementation project(path: ":sdks:java:io:iceberg")
testImplementation project(path: ":it:common", configuration:
"testRuntimeMigration")
testImplementation library.java.mockito_inline
testImplementation project(path:
":sdks:java:extensions:google-cloud-platform-core", configuration:
"testRuntimeMigration")
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 3f486ec3f5d..b4e477f8ec0 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,7 +1,358 @@
+/*
+ * 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.
+ */
package org.apache.beam.it.iceberg;
+import static
org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatResult;
+import static org.junit.Assert.assertEquals;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.text.ParseException;
+import java.time.Duration;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.Map;
+import java.util.Random;
+import java.util.UUID;
+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.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
import org.apache.beam.it.common.dataflow.IOLoadTestBase;
+import org.apache.beam.it.common.storage.GcsResourceManager;
+import org.apache.beam.it.common.utils.ResourceManagerUtils;
+import org.apache.beam.sdk.io.GenerateSequence;
+import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig;
+import org.apache.beam.sdk.io.iceberg.IcebergIO;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.TestPipelineOptions;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+
+/**
+ * IcebergIO performance tests using a Hadoop catalog backed by GCS.
+ *
+ * <p>The test writes generated rows to an Iceberg table, reads the table
back, validates the number
+ * of rows, and exports write and read performance metrics.
+ *
+ * <p>Example trigger command:
+ *
+ * <pre>
+ * ./gradlew :it:iceberg:integrationTest \
+ * --tests "org.apache.beam.it.iceberg.IcebergIOLT" \
+ * -Dconfiguration=local \
+ * -Dproject=[gcpProject] \
+ * -DartifactBucket=[artifactBucket]
+ * </pre>
+ */
+public final class IcebergIOLT extends IOLoadTestBase {
+
+ private static final String READ_ELEMENT_METRIC_NAME = "read_count";
+ private static final String NAMESPACE = "default";
+
+ private static final Schema ROW_SCHEMA =
+ Schema.builder().addInt64Field("id").addByteArrayField("data").build();
+
+ private static final Map<String, Configuration> TEST_CONFIGS =
+ ImmutableMap.of(
+ "local",
+ Configuration.of(1_000L, 5, "DirectRunner", 1_000), // approximately
1 MB
+ "small",
+ Configuration.of(1_000_000L, 20, "DataflowRunner", 1_000), //
approximately 1 GB
+ "medium",
+ Configuration.of(10_000_000L, 40, "DataflowRunner", 1_000), //
approximately 10 GB
+ "large",
+ Configuration.of(100_000_000L, 100, "DataflowRunner", 1_000) //
approximately 100 GB
+ );
+
+ private static GcsResourceManager resourceManager;
+
+ private Configuration configuration;
+ private IcebergCatalogConfig catalogConfig;
+ private TableIdentifier tableIdentifier;
+
+ @Rule public TestPipeline writePipeline = TestPipeline.create();
+
+ @Rule public TestPipeline readPipeline = TestPipeline.create();
+
+ @BeforeClass
+ public static void beforeClass() {
+ resourceManager =
+ GcsResourceManager.builder(TestProperties.artifactBucket(),
"icebergiolt", CREDENTIALS)
+ .build();
+ }
+
+ @AfterClass
+ public static void tearDownClass() {
+ ResourceManagerUtils.cleanResources(resourceManager);
+ }
+
+ @Before
+ public void setup() {
+ configuration = getTestConfiguration();
+
+ String uniqueSuffix =
+ DateTimeFormatter.ofPattern("MMddHHmmssSSS")
+ .withZone(ZoneOffset.UTC)
+ .format(java.time.Instant.now())
+ + "-"
+ + UUID.randomUUID().toString().substring(0, 10);
+
+ String warehouseDirectory = "icebergiolt-" + uniqueSuffix;
+ resourceManager.registerTempDir(warehouseDirectory);
+
+ String warehouseLocation =
+ String.format("gs://%s/%s", TestProperties.artifactBucket(),
warehouseDirectory);
+
+ tableIdentifier = TableIdentifier.of(NAMESPACE, "table_" +
uniqueSuffix.replace("-", "_"));
+
+ Map<String, String> catalogProperties =
+ ImmutableMap.of(
+ "type", CatalogUtil.ICEBERG_CATALOG_TYPE_HADOOP, "warehouse",
warehouseLocation);
+
+ catalogConfig =
+ IcebergCatalogConfig.builder()
+ .setCatalogName("iceberg-load-test")
+ .setCatalogProperties(catalogProperties)
+ .build();
+
+ setTempLocation(writePipeline);
+ setTempLocation(readPipeline);
+ }
+
+ /** Writes generated rows to Iceberg and verifies that all rows can be read
back. */
+ @Test
+ public void testIcebergWriteAndRead() throws IOException {
+ PipelineLauncher.LaunchInfo writeInfo = testWrite();
+
+ PipelineOperator.Result writeResult =
+ pipelineOperator.waitUntilDone(
+ createConfig(writeInfo,
Duration.ofMinutes(configuration.getPipelineTimeout())));
+
+ assertThatResult(writeResult).isLaunchFinished();
+ assertEquals(
+ PipelineLauncher.JobState.DONE,
+ pipelineLauncher.getJobStatus(project, region, writeInfo.jobId()));
+
+ PipelineLauncher.LaunchInfo readInfo = testRead();
+
+ PipelineOperator.Result readResult =
+ pipelineOperator.waitUntilDone(
+ createConfig(readInfo,
Duration.ofMinutes(configuration.getPipelineTimeout())));
+
+ assertThatResult(readResult).isLaunchFinished();
+ assertEquals(
+ PipelineLauncher.JobState.DONE,
+ pipelineLauncher.getJobStatus(project, region, readInfo.jobId()));
+
+ double numRecords =
+ pipelineLauncher.getMetric(
+ project,
+ region,
+ readInfo.jobId(),
+ getBeamMetricsName(PipelineMetricsType.COUNTER,
READ_ELEMENT_METRIC_NAME));
+
+ assertEquals((double) configuration.getNumRows(), numRecords, 0.5);
+
+ exportMetrics(writeInfo, readInfo);
+ }
+
+ private PipelineLauncher.LaunchInfo testWrite() throws IOException {
+ writePipeline
+ .apply("Generate records",
GenerateSequence.from(0).to(configuration.getNumRows()))
+ .apply("Map records", ParDo.of(new
MapToIcebergFormat(configuration.getValueSizeBytes())))
+ .setRowSchema(ROW_SCHEMA)
+ .apply("Write to Iceberg",
IcebergIO.writeRows(catalogConfig).to(tableIdentifier));
+
+ PipelineLauncher.LaunchConfig options =
+ PipelineLauncher.LaunchConfig.builder("write-iceberg")
+ .setSdk(PipelineLauncher.Sdk.JAVA)
+ .setPipeline(writePipeline)
+ .addParameter("runner", configuration.getRunner())
+ .addParameter(
+ "maxNumWorkers",
+ TestProperties.getProperty("maxNumWorkers", "10",
TestProperties.Type.PROPERTY))
+ .build();
+
+ return pipelineLauncher.launch(project, region, options);
+ }
+
+ private PipelineLauncher.LaunchInfo testRead() throws IOException {
+ readPipeline
+ .apply("Read from Iceberg",
IcebergIO.readRows(catalogConfig).from(tableIdentifier))
+ .apply("Counting element", ParDo.of(new
CountingFn<>(READ_ELEMENT_METRIC_NAME)));
+
+ PipelineLauncher.LaunchConfig options =
+ PipelineLauncher.LaunchConfig.builder("read-iceberg")
+ .setSdk(PipelineLauncher.Sdk.JAVA)
+ .setPipeline(readPipeline)
+ .addParameter("runner", configuration.getRunner())
+ .addParameter(
+ "maxNumWorkers",
+ TestProperties.getProperty("maxNumWorkers", "10",
TestProperties.Type.PROPERTY))
+ .build();
+
+ return pipelineLauncher.launch(project, region, options);
+ }
+
+ private void exportMetrics(
+ PipelineLauncher.LaunchInfo writeInfo, PipelineLauncher.LaunchInfo
readInfo) {
+
+ /*
+ * Only use PCollection names belonging to transforms defined directly in
this test.
+ * Internal IcebergIO transform names may change when the implementation
changes.
+ */
+ MetricsConfiguration writeMetricsConfig =
+ MetricsConfiguration.builder()
+ .setInputPCollection("Map records.out0")
+ .setInputPCollectionV2("Map
records/ParMultiDo(MapToIcebergFormat).out0")
+ .build();
+
+ MetricsConfiguration readMetricsConfig =
+ MetricsConfiguration.builder()
+ .setOutputPCollection("Counting element.out0")
+ .setOutputPCollectionV2("Counting
element/ParMultiDo(Counting).out0")
+ .build();
+
+ try {
+ exportMetricsToBigQuery(writeInfo, getMetrics(writeInfo,
writeMetricsConfig));
+
+ exportMetricsToBigQuery(readInfo, getMetrics(readInfo,
readMetricsConfig));
+ } catch (IOException | ParseException | InterruptedException e) {
+ throw new RuntimeException("Unable to collect or export IcebergIO
load-test metrics.", e);
+ }
+ }
+
+ private Configuration getTestConfiguration() {
+ String testConfig =
+ TestProperties.getProperty("configuration", "local",
TestProperties.Type.PROPERTY);
+
+ Configuration selectedConfiguration = TEST_CONFIGS.get(testConfig);
+
+ if (selectedConfiguration == null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unknown test configuration: [%s]. Known configurations: %s",
+ testConfig, TEST_CONFIGS.keySet()));
+ }
+
+ return selectedConfiguration;
+ }
+
+ private static void setTempLocation(TestPipeline pipeline) {
+ String tempBucketName = TestProperties.artifactBucket();
+
+ if (Strings.isNullOrEmpty(tempBucketName)) {
+ return;
+ }
+
+ String tempLocation = String.format("gs://%s/temp/", tempBucketName);
+
+
pipeline.getOptions().as(TestPipelineOptions.class).setTempRoot(tempLocation);
+
+ pipeline.getOptions().setTempLocation(tempLocation);
+ }
+
+ /** Maps a generated sequence value to a deterministic Iceberg-compatible
Beam row. */
+ private static class MapToIcebergFormat extends DoFn<Long, Row> {
+
+ private final int valueSizeBytes;
+
+ private MapToIcebergFormat(int valueSizeBytes) {
+ if (valueSizeBytes <= 0) {
+ throw new IllegalArgumentException("valueSizeBytes must be positive.");
+ }
+
+ this.valueSizeBytes = valueSizeBytes;
+ }
+
+ @ProcessElement
+ public void processElement(@Element Long index, OutputReceiver<Row>
output) {
+
+ byte[] value = new byte[valueSizeBytes];
+
+ /*
+ * Using the row index as the seed makes retries deterministic. Random
data also prevents
+ * compression from making the generated payload significantly smaller
than configured.
+ */
+ new Random(index).nextBytes(value);
+
+ output.output(Row.withSchema(ROW_SCHEMA).addValues(index,
value).build());
+ }
+ }
+
+ /** Options for the IcebergIO load test. */
+ @AutoValue
+ abstract static class Configuration {
+
+ abstract long getNumRows();
+
+ abstract int getPipelineTimeout();
+
+ abstract String getRunner();
+
+ abstract int getValueSizeBytes();
+
+ static Configuration of(long numRows, int pipelineTimeout, String runner,
int valueSizeBytes) {
+
+ if (numRows <= 0) {
+ throw new IllegalArgumentException("numRows must be positive.");
+ }
+
+ if (pipelineTimeout <= 0) {
+ throw new IllegalArgumentException("pipelineTimeout must be
positive.");
+ }
+
+ if (valueSizeBytes <= 0) {
+ throw new IllegalArgumentException("valueSizeBytes must be positive.");
+ }
+
+ return new AutoValue_IcebergIOLT_Configuration.Builder()
+ .setNumRows(numRows)
+ .setPipelineTimeout(pipelineTimeout)
+ .setRunner(runner)
+ .setValueSizeBytes(valueSizeBytes)
+ .build();
+ }
+
+ @AutoValue.Builder
+ abstract static class Builder {
+
+ abstract Builder setNumRows(long numRows);
+
+ abstract Builder setPipelineTimeout(int pipelineTimeout);
+
+ abstract Builder setRunner(String runner);
-public class IcebergIOLT extends IOLoadTestBase {
+ abstract Builder setValueSizeBytes(int valueSizeBytes);
+ abstract Configuration build();
+ }
+ }
}
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 4b0ecbfd343..1607ca1cd67 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,12 +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.common.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
import org.apache.beam.it.common.dataflow.IOLoadTestBase;
+import org.apache.beam.it.common.utils.ResourceManagerUtils;
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 cfc11127c34..d6231a69f1e 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,10 @@ 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.common.dataflow.DefaultPipelineLauncher.PipelineMetricsType;
import org.apache.beam.it.common.dataflow.IOStressTestBase;
import
org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions;
import org.apache.beam.sdk.io.Read;
