You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by vt...@apache.org on 2026/07/09 10:44:17 UTC
(beam) 02/03: Add IcebergIOLT
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 5312b03502c4bd02e77fe8293f2fb31766d8ff35
Author: Vitaly Terentyev <vi...@akvelon.com>
AuthorDate: Tue Jul 7 17:04:31 2026 +0400
Add IcebergIOLT
---
it/clickhouse/build.gradle | 2 +-
it/common/build.gradle | 23 +-
.../apache/beam/it/common}/artifacts/Artifact.java | 2 +-
.../beam/it/common}/artifacts/ArtifactClient.java | 2 +-
.../beam/it/common}/artifacts/GcsArtifact.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 +-
it/common/src/main/resources/test-artifact.json | 1 +
.../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 +-
it/google-cloud-platform/build.gradle | 18 --
.../it/gcp/artifacts/matchers/package-info.java | 20 --
.../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 | 42 +--
.../org/apache/beam/it/iceberg/IcebergIOLT.java | 351 ++++++++++++++++++++-
.../java/org/apache/beam/it/kafka/KafkaIOLT.java | 4 +-
.../java/org/apache/beam/it/kafka/KafkaIOST.java | 2 +-
it/truthmatchers/build.gradle | 6 +-
.../beam/it/truthmatchers}/ArtifactAsserts.java | 15 +-
.../beam/it/truthmatchers}/ArtifactsSubject.java | 21 +-
45 files changed, 459 insertions(+), 174 deletions(-)
diff --git a/it/clickhouse/build.gradle b/it/clickhouse/build.gradle
index 858d0dea920..9fa1c8634a9 100644
--- a/it/clickhouse/build.gradle
+++ b/it/clickhouse/build.gradle
@@ -52,4 +52,4 @@ dependencies {
permitUnusedDeclared "com.clickhouse:client-v2:$clickhouse_java_client_version:all"
-}
\ No newline at end of file
+}
diff --git a/it/common/build.gradle b/it/common/build.gradle
index 7b8559aa6e1..62dd45ddacf 100644
--- a/it/common/build.gradle
+++ b/it/common/build.gradle
@@ -29,31 +29,40 @@ dependencies {
implementation enforcedPlatform(library.java.google_cloud_platform_libraries_bom)
implementation project(path: ":sdks:java:core", configuration: "shadow")
implementation project(path: ":runners:google-cloud-dataflow-java")
- 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'
+ implementation 'org.apache.hadoop:hadoop-common:3.3.6'
+ implementation 'org.apache.avro:avro:1.11.1'
+ implementation 'org.apache.parquet:parquet-avro:1.15.2'
+ implementation 'org.apache.parquet:parquet-common:1.15.2'
+ implementation 'org.apache.parquet:parquet-hadoop:1.15.2'
provided 'com.google.api.grpc:proto-google-cloud-monitoring-v3'
+ implementation library.java.jackson_core
+ implementation library.java.jackson_databind
implementation library.java.gax
implementation library.java.google_api_services_dataflow
implementation library.java.google_auth_library_credentials
implementation library.java.google_auth_library_oauth2_http
- implementation library.java.google_api_services_bigquery
implementation library.java.vendored_guava_32_1_2_jre
implementation library.java.slf4j_api
implementation library.java.commons_lang3
implementation library.java.failsafe
implementation library.java.google_code_gson
+ implementation library.java.google_cloud_core
+ implementation library.java.google_api_client
implementation library.java.google_http_client
implementation library.java.guava
implementation library.java.protobuf_java_util
implementation library.java.protobuf_java
-
+ implementation library.java.junit
+ testImplementation library.java.mockito_inline
+ testRuntimeOnly library.java.slf4j_simple
// TODO: excluding Guava until Truth updates it to >32.1.x
testImplementation(library.java.truth) {
- exclude group: 'com.google.guava', module: 'guava'
+ exclude group: 'com.google.guava', module: 'guava'
}
- testImplementation library.java.junit
- testImplementation library.java.mockito_inline
- testRuntimeOnly library.java.slf4j_simple
+ testImplementation project(path: ":runners:direct-java", configuration: "shadow")
+ testImplementation project(path: ":sdks:java:testing:test-utils")
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/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/common/src/main/resources/test-artifact.json b/it/common/src/main/resources/test-artifact.json
new file mode 100644
index 00000000000..551c80d14a6
--- /dev/null
+++ b/it/common/src/main/resources/test-artifact.json
@@ -0,0 +1 @@
+["This is a test artifact."]
\ No newline at end of file
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/build.gradle b/it/google-cloud-platform/build.gradle
index dd8376e3569..164a75c06ba 100644
--- a/it/google-cloud-platform/build.gradle
+++ b/it/google-cloud-platform/build.gradle
@@ -31,47 +31,29 @@ evaluationDependsOn(":it:common")
dependencies {
implementation enforcedPlatform(library.java.google_cloud_platform_libraries_bom)
- implementation project(path: ":sdks:java:core", configuration: "shadow")
- implementation project(path: ":runners:google-cloud-dataflow-java")
implementation project(path: ":it:common")
implementation project(path: ":it:conditions")
implementation project(path: ":it:truthmatchers")
- implementation project(path: ":sdks:java:testing:test-utils")
implementation library.java.failsafe
implementation library.java.google_api_services_dataflow
implementation library.java.junit
implementation library.java.slf4j_api
- implementation(library.java.truth) {
- exclude group: 'com.google.guava', module: 'guava'
- }
implementation library.java.vendored_guava_32_1_2_jre
implementation library.java.jackson_core
implementation library.java.jackson_databind
- implementation 'org.apache.hadoop:hadoop-common:3.3.5'
- implementation 'org.apache.avro:avro:1.11.1'
- implementation 'org.apache.parquet:parquet-avro:1.15.2'
- implementation 'org.apache.parquet:parquet-common:1.15.2'
- implementation 'org.apache.parquet:parquet-hadoop:1.15.2'
implementation 'org.apache.commons:commons-lang3:3.9'
implementation library.java.gax
implementation library.java.google_api_common
implementation library.java.google_api_client
- implementation library.java.google_api_services_bigquery
implementation library.java.google_auth_library_credentials
implementation library.java.google_auth_library_oauth2_http
- implementation library.java.google_http_client
implementation library.java.guava
- implementation library.java.protobuf_java_util
implementation library.java.protobuf_java
implementation library.java.threetenbp
implementation 'org.awaitility:awaitility:4.2.0'
- implementation 'joda-time:joda-time:2.10.10'
// Google Cloud Dependencies
implementation library.java.google_cloud_core
- implementation 'com.google.cloud:google-cloud-storage'
implementation 'com.google.cloud:google-cloud-bigquery'
- implementation 'com.google.cloud:google-cloud-monitoring'
- provided 'com.google.api.grpc:proto-google-cloud-monitoring-v3'
implementation 'com.google.cloud:google-cloud-bigtable'
implementation 'com.google.cloud:google-cloud-spanner'
implementation 'com.google.cloud:google-cloud-pubsub'
diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/matchers/package-info.java b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/matchers/package-info.java
deleted file mode 100644
index 9759071b397..00000000000
--- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/matchers/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.
- */
-
-/** Package for Artifact Truth matchers / subjects to have reusable assertions. */
-package org.apache.beam.it.gcp.artifacts.matchers;
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..f11374a1ca4 100644
--- a/it/iceberg/build.gradle
+++ b/it/iceberg/build.gradle
@@ -31,7 +31,6 @@ ext.summary = "Integration test utilities for Iceberg."
def iceberg_version = "1.10.0"
def parquet_version = "1.16.0"
-def orc_version = "1.9.6"
def hive_version = "3.1.3"
def hadoopVersions = [
@@ -43,27 +42,8 @@ hadoopVersions.each {kv -> configurations.create("hadoopVersion$kv.key")}
evaluationDependsOn(":it:common")
dependencies {
- implementation library.java.vendored_guava_32_1_2_jre
- implementation project(path: ":sdks:java:core", configuration: "shadow")
- implementation project(path: ":model:pipeline", configuration: "shadow")
- implementation library.java.avro
- implementation library.java.slf4j_api
- implementation library.java.joda_time
- implementation "org.apache.parquet:parquet-column:$parquet_version"
- implementation "org.apache.parquet:parquet-hadoop:$parquet_version"
- implementation "org.apache.parquet:parquet-common:$parquet_version"
- implementation project(":sdks:java:io:parquet")
- implementation "org.apache.orc:orc-core:$orc_version"
- implementation "org.apache.iceberg:iceberg-core:$iceberg_version"
- implementation "org.apache.iceberg:iceberg-api:$iceberg_version"
- implementation "org.apache.iceberg:iceberg-parquet:$iceberg_version"
- implementation "org.apache.iceberg:iceberg-orc:$iceberg_version"
- implementation "org.apache.iceberg:iceberg-data:$iceberg_version"
- implementation "org.apache.hadoop:hadoop-common:3.3.6"
-
provided "org.immutables:value:2.8.8"
permitUnusedDeclared "org.immutables:value:2.8.8"
- implementation library.java.vendored_calcite_1_40_0
runtimeOnly "org.apache.iceberg:iceberg-gcp:$iceberg_version"
runtimeOnly "org.apache.iceberg:iceberg-aws:$iceberg_version"
runtimeOnly "org.apache.iceberg:iceberg-aws-bundle:$iceberg_version"
@@ -113,28 +93,10 @@ dependencies {
"hadoopVersion$kv.key" "org.apache.hadoop:hadoop-mapreduce-client-core:$kv.value"
}
- implementation project(path: ":runners:google-cloud-dataflow-java")
- implementation project(path: ":it:common")
- implementation project(path: ":it:conditions")
- implementation project(path: ":it:truthmatchers")
- implementation project(path: ":sdks:java:testing:test-utils")
- implementation library.java.failsafe
- implementation library.java.google_api_services_dataflow
- implementation(library.java.truth) {
- exclude group: 'com.google.guava', module: 'guava'
- }
- implementation library.java.jackson_core
- implementation library.java.jackson_databind
- implementation 'org.apache.commons:commons-lang3:3.9'
- implementation library.java.google_api_client
- implementation library.java.google_http_client
- implementation library.java.guava
- implementation library.java.protobuf_java_util
- implementation library.java.protobuf_java
- implementation library.java.threetenbp
-
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:truthmatchers")
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..683ff4529b6 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,356 @@
+/*
+ * 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.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", "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;
diff --git a/it/truthmatchers/build.gradle b/it/truthmatchers/build.gradle
index fd54ad591a0..2e21c1b5df0 100644
--- a/it/truthmatchers/build.gradle
+++ b/it/truthmatchers/build.gradle
@@ -27,11 +27,15 @@ ext.summary = "Truth matchers for integration tests."
dependencies {
implementation project(path: ":it:common")
+ implementation 'org.apache.avro:avro:1.11.1'
// TODO: excluding Guava until Truth updates it to >32.1.x
implementation(library.java.truth) {
exclude group: 'com.google.guava', module: 'guava'
}
+ implementation library.java.jackson_core
+ implementation library.java.jackson_databind
implementation library.java.guava
+ implementation library.java.vendored_guava_32_1_2_jre
permitUsedUndeclared library.java.guava
implementation library.java.google_code_gson
-}
\ No newline at end of file
+}
diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/matchers/ArtifactAsserts.java b/it/truthmatchers/src/main/java/org/apache/beam/it/truthmatchers/ArtifactAsserts.java
similarity index 79%
rename from it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/matchers/ArtifactAsserts.java
rename to it/truthmatchers/src/main/java/org/apache/beam/it/truthmatchers/ArtifactAsserts.java
index 3b1251307d0..92407e5a2fc 100644
--- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/matchers/ArtifactAsserts.java
+++ b/it/truthmatchers/src/main/java/org/apache/beam/it/truthmatchers/ArtifactAsserts.java
@@ -15,15 +15,12 @@
* 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.truthmatchers;
+import com.google.common.truth.Truth;
import java.util.List;
import org.apache.avro.generic.GenericRecord;
-import org.apache.beam.it.gcp.artifacts.Artifact;
-import org.apache.beam.it.truthmatchers.RecordsSubject;
+import org.apache.beam.it.common.artifacts.Artifact;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
public class ArtifactAsserts {
@@ -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/truthmatchers/src/main/java/org/apache/beam/it/truthmatchers/ArtifactsSubject.java
similarity index 87%
rename from it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/matchers/ArtifactsSubject.java
rename to it/truthmatchers/src/main/java/org/apache/beam/it/truthmatchers/ArtifactsSubject.java
index ab5b4969948..e2518280345 100644
--- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/artifacts/matchers/ArtifactsSubject.java
+++ b/it/truthmatchers/src/main/java/org/apache/beam/it/truthmatchers/ArtifactsSubject.java
@@ -15,11 +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.truthmatchers;
-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.vendor.guava.v32_1_2_jre.com.google.common.hash.Hashing.sha256;
+import static org.apache.beam.it.truthmatchers.ArtifactAsserts.assertThatGenericRecords;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -31,11 +29,11 @@ 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.truthmatchers.RecordsSubject;
+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.vendor.guava.v32_1_2_jre.com.google.common.hash.Hashing;
/**
* Subject that has assertion operations for artifact lists (GCS files), usually coming from the
@@ -98,7 +96,8 @@ public final class ArtifactsSubject extends Subject {
*/
public void hasHash(String hash) {
if (actual.stream()
- .noneMatch(artifact -> sha256().hashBytes(artifact.contents()).toString().equals(hash))) {
+ .noneMatch(
+ artifact -> Hashing.sha256().hashBytes(artifact.contents()).toString().equals(hash))) {
failWithActual("expected to contain hash", hash);
}
}
@@ -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);
}
}