diff --git a/.github/workflows/beam_PostCommit_Java_IO_Performance_Tests.yml b/.github/workflows/beam_PostCommit_Java_IO_Performance_Tests.yml index db43441f2cda..9bda3d1bf31f 100644 --- a/.github/workflows/beam_PostCommit_Java_IO_Performance_Tests.yml +++ b/.github/workflows/beam_PostCommit_Java_IO_Performance_Tests.yml @@ -64,7 +64,7 @@ jobs: matrix: job_name: ["beam_PostCommit_Java_IO_Performance_Tests"] job_phrase: ["Run Java PostCommit IO Performance Tests"] - test_case: ["GCSPerformanceTest", "BigTablePerformanceTest", "BigQueryStorageApiStreamingPerformanceTest"] + test_case: ["IcebergPerformanceTest", "GCSPerformanceTest", "BigTablePerformanceTest", "BigQueryStorageApiStreamingPerformanceTest"] steps: - uses: actions/checkout@v7 - name: Setup repository diff --git a/.test-infra/tools/refresh_looker_metrics.py b/.test-infra/tools/refresh_looker_metrics.py index c8d66f4a4bd3..573cee0d4e2b 100644 --- a/.test-infra/tools/refresh_looker_metrics.py +++ b/.test-infra/tools/refresh_looker_metrics.py @@ -34,6 +34,8 @@ ("33", ["21", "70", "116", "69", "115"]), # BigTableIO_Write ("34", ["22", "56", "96", "55", "95"]), # TextIO_Read ("35", ["23", "64", "110", "63", "109"]), # TextIO_Write + ("113", ["386", "388", "390", "392", "394"]), # IcebergIO_Read + ("114", ["387", "389", "391", "393", "395"]), # IcebergIO_Write ("75", ["258", "259", "260", "261", "262"]), # TensorFlow MNIST ("76", ["233", "234", "235", "236", "237"]), # PyTorch BERT base uncased ("77", ["238", "239", "240", "241", "242"]), # PyTorch BERT large uncased diff --git a/it/build.gradle b/it/build.gradle index 42a9ad9f4ee8..c550e26d305f 100644 --- a/it/build.gradle +++ b/it/build.gradle @@ -34,4 +34,8 @@ tasks.register('BigTablePerformanceTest') { tasks.register('BigQueryStorageApiStreamingPerformanceTest') { dependsOn(":it:google-cloud-platform:BigQueryStorageApiStreamingPerformanceTest") -} \ No newline at end of file +} + +tasks.register('IcebergPerformanceTest') { + dependsOn(":it:iceberg:IcebergPerformanceTest") +} diff --git a/it/clickhouse/build.gradle b/it/clickhouse/build.gradle index 858d0dea9207..0727f0328e45 100644 --- a/it/clickhouse/build.gradle +++ b/it/clickhouse/build.gradle @@ -29,6 +29,12 @@ ext.summary = "Integration test utilities for ClickHouse." def clickhouse_java_client_version = "0.9.6" +configurations.configureEach { + resolutionStrategy.capabilitiesResolution.withCapability("org.lz4:lz4-java") { + select("at.yawk.lz4:lz4-java:1.10.2") + } +} + dependencies { implementation project(path: ":it:common") implementation project(path: ":it:testcontainers") @@ -52,4 +58,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 c92ec551cb25..62dd45ddacf5 100644 --- a/it/common/build.gradle +++ b/it/common/build.gradle @@ -28,6 +28,19 @@ ext.summary = "Code used by all integration test utilities." 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 '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 @@ -36,14 +49,20 @@ dependencies { 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 830f44598be7..6371d6eedb1e 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 e463baffeaf1..b275d35703e1 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 c60e3a0fe53a..5f4360ddbb2b 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 de2a2541d558..44b54f82aeba 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 854651ae1f14..3a3bcb2d7ec4 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 5cfe4c137e35..58241f9a1db3 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 9a83558f7bfc..08c3cdfb0c5a 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 d6ebf825e2c1..fb4326e1b4a7 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 53cb4ed1d3af..aaf820b25103 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/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManager.java b/it/common/src/main/java/org/apache/beam/it/common/bigquery/BigQueryResourceManager.java similarity index 99% rename from it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManager.java rename to it/common/src/main/java/org/apache/beam/it/common/bigquery/BigQueryResourceManager.java index 5ca6a8c4655c..728a47b224b4 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManager.java +++ b/it/common/src/main/java/org/apache/beam/it/common/bigquery/BigQueryResourceManager.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.it.gcp.bigquery; +package org.apache.beam.it.common.bigquery; import com.google.api.gax.paging.Page; import com.google.auth.Credentials; diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerException.java b/it/common/src/main/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerException.java similarity index 96% rename from it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerException.java rename to it/common/src/main/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerException.java index 61ef27e45260..ce95dc891551 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerException.java +++ b/it/common/src/main/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerException.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.it.gcp.bigquery; +package org.apache.beam.it.common.bigquery; /** Custom exception for {@link BigQueryResourceManager} implementations. */ public class BigQueryResourceManagerException extends RuntimeException { diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerUtils.java b/it/common/src/main/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerUtils.java similarity index 98% rename from it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerUtils.java rename to it/common/src/main/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerUtils.java index f2b6849caa61..56cb82013885 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerUtils.java +++ b/it/common/src/main/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerUtils.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.it.gcp.bigquery; +package org.apache.beam.it.common.bigquery; import static org.apache.beam.it.common.utils.ResourceManagerUtils.generateResourceId; diff --git a/it/common/src/main/java/org/apache/beam/it/common/bigquery/package-info.java b/it/common/src/main/java/org/apache/beam/it/common/bigquery/package-info.java new file mode 100644 index 000000000000..f2d5978fb4eb --- /dev/null +++ b/it/common/src/main/java/org/apache/beam/it/common/bigquery/package-info.java @@ -0,0 +1,20 @@ +/* + * 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 managing BigQuery common resources within integration tests. */ +package org.apache.beam.it.common.bigquery; diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/AbstractPipelineLauncher.java b/it/common/src/main/java/org/apache/beam/it/common/dataflow/AbstractPipelineLauncher.java similarity index 99% rename from it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/AbstractPipelineLauncher.java rename to it/common/src/main/java/org/apache/beam/it/common/dataflow/AbstractPipelineLauncher.java index 9d00fa67f145..a6efaeca5afc 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/AbstractPipelineLauncher.java +++ b/it/common/src/main/java/org/apache/beam/it/common/dataflow/AbstractPipelineLauncher.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.it.gcp.dataflow; +package org.apache.beam.it.common.dataflow; import static org.apache.beam.it.common.PipelineLauncher.JobState.ACTIVE_STATES; import static org.apache.beam.it.common.PipelineLauncher.JobState.FAILED; diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DefaultPipelineLauncher.java b/it/common/src/main/java/org/apache/beam/it/common/dataflow/DefaultPipelineLauncher.java similarity index 92% rename from it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DefaultPipelineLauncher.java rename to it/common/src/main/java/org/apache/beam/it/common/dataflow/DefaultPipelineLauncher.java index 11a09c4ba749..af6438a561aa 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DefaultPipelineLauncher.java +++ b/it/common/src/main/java/org/apache/beam/it/common/dataflow/DefaultPipelineLauncher.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.dataflow; +package org.apache.beam.it.common.dataflow; import static org.apache.beam.it.common.logging.LogStrings.formatForLogging; -import static org.apache.beam.sdk.testing.TestPipeline.PROPERTY_BEAM_TEST_PIPELINE_OPTIONS; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; import com.fasterxml.jackson.databind.ObjectMapper; @@ -41,7 +40,6 @@ import java.util.stream.StreamSupport; import org.apache.beam.it.common.PipelineLauncher; import org.apache.beam.it.common.utils.PipelineUtils; -import org.apache.beam.it.gcp.IOLoadTestBase; import org.apache.beam.runners.dataflow.DataflowPipelineJob; import org.apache.beam.sdk.PipelineResult; import org.apache.beam.sdk.metrics.DistributionResult; @@ -73,6 +71,8 @@ public class DefaultPipelineLauncher extends AbstractPipelineLauncher { private static final String READ_PIPELINE_NAME_OVERWRITE = "readPipelineNameOverride"; private static final String WRITE_PIPELINE_NAME_OVERWRITE = "writePipelineNameOverride"; private static final Pattern JOB_ID_PATTERN = Pattern.compile("Submitted job: (\\S+)"); + /** Namespace for Beam provided pipeline metrics (set up by Metrics transform). */ + public static final String BEAM_METRICS_NAMESPACE = "BEAM_METRICS"; // For unsupported runners (other than dataflow), implement launcher methods by operating with // PipelineResult. @@ -101,6 +101,18 @@ public class DefaultPipelineLauncher extends AbstractPipelineLauncher { .put(PipelineResult.State.UNRECOGNIZED, JobState.UNKNOWN) .build(); + // To make PipelineLauncher.getMetric work in a unified way for both runner provided metrics and + // pipeline defined + // metrics, here we wrap Beam provided metrics as a pre-defined metrics name + // [name_space:metric_type:metric_name + // which will be recognized by getMetric method + public enum PipelineMetricsType { + COUNTER, + STARTTIME, + ENDTIME, + RUNTIME, + } + private DefaultPipelineLauncher(Builder builder) { super( new Dataflow( @@ -169,7 +181,7 @@ private static void checkIfMetricResultIsUnique( resultCount <= 1, "More than one metric result matches name: %s in namespace %s. Metric results count: %s", name, - IOLoadTestBase.BEAM_METRICS_NAMESPACE, + BEAM_METRICS_NAMESPACE, resultCount); } @@ -180,15 +192,13 @@ private static Iterable> getDistributions( .metrics() .queryMetrics( MetricsFilter.builder() - .addNameFilter( - MetricNameFilter.named(IOLoadTestBase.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, IOLoadTestBase.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) { @@ -197,8 +207,7 @@ public Long getBeamMetric( .metrics() .queryMetrics( MetricsFilter.builder() - .addNameFilter( - MetricNameFilter.named(IOLoadTestBase.BEAM_METRICS_NAMESPACE, metricName)) + .addNameFilter(MetricNameFilter.named(BEAM_METRICS_NAMESPACE, metricName)) .build()); switch (metricType) { @@ -210,9 +219,7 @@ public Long getBeamMetric( return metricResult.getAttempted(); } catch (NoSuchElementException e) { LOG.error( - "Failed to get metric {}, from namespace {}", - metricName, - IOLoadTestBase.BEAM_METRICS_NAMESPACE); + "Failed to get metric {}, from namespace {}", metricName, BEAM_METRICS_NAMESPACE); } return UNKNOWN_METRIC_VALUE; case STARTTIME: @@ -230,9 +237,9 @@ public Long getBeamMetric( .map(element -> Objects.requireNonNull(element.getAttempted()).getMax()) .max(Long::compareTo) .orElse(UNKNOWN_METRIC_VALUE); - if (metricType == IOLoadTestBase.PipelineMetricsType.STARTTIME) { + if (metricType == PipelineMetricsType.STARTTIME) { return lowestMin; - } else if (metricType == IOLoadTestBase.PipelineMetricsType.ENDTIME) { + } else if (metricType == PipelineMetricsType.ENDTIME) { return greatestMax; } else { if (lowestMin != UNKNOWN_METRIC_VALUE && greatestMax != UNKNOWN_METRIC_VALUE) { @@ -254,15 +261,14 @@ public Long getBeamMetric( @Override public Double getMetric(String project, String region, String jobId, String metricName) throws IOException { - if (metricName.startsWith(IOLoadTestBase.BEAM_METRICS_NAMESPACE)) { + if (metricName.startsWith(BEAM_METRICS_NAMESPACE)) { String[] nameSpacedMetrics = metricName.split(":", 3); Preconditions.checkState( nameSpacedMetrics.length == 3, String.format( "Invalid Beam metrics name: %s, expected: '%s:metric_type:metric_name'", - metricName, IOLoadTestBase.BEAM_METRICS_NAMESPACE)); - IOLoadTestBase.PipelineMetricsType metricType = - IOLoadTestBase.PipelineMetricsType.valueOf(nameSpacedMetrics[1]); + metricName, BEAM_METRICS_NAMESPACE)); + PipelineMetricsType metricType = PipelineMetricsType.valueOf(nameSpacedMetrics[1]); // Pipeline defined metrics are long values. Have to cast to double that is what the base // class defined. @@ -437,16 +443,18 @@ private List extractOptions(String project, String region, LaunchConfig // add pipeline options from beamTestPipelineOptions system property to preserve the // pipeline options already set in TestPipeline. @Nullable - String beamTestPipelineOptions = System.getProperty(PROPERTY_BEAM_TEST_PIPELINE_OPTIONS); + String beamTestPipelineOptions = + System.getProperty( + org.apache.beam.sdk.testing.TestPipeline.PROPERTY_BEAM_TEST_PIPELINE_OPTIONS); if (!Strings.isNullOrEmpty(beamTestPipelineOptions)) { try { additionalOptions.addAll(MAPPER.readValue(beamTestPipelineOptions, List.class)); } catch (IOException e) { throw new RuntimeException( "Unable to instantiate test options from system property " - + PROPERTY_BEAM_TEST_PIPELINE_OPTIONS + + org.apache.beam.sdk.testing.TestPipeline.PROPERTY_BEAM_TEST_PIPELINE_OPTIONS + ":" - + System.getProperty(PROPERTY_BEAM_TEST_PIPELINE_OPTIONS), + + beamTestPipelineOptions, e); } } diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datagenerator/package-info.java b/it/common/src/main/java/org/apache/beam/it/common/dataflow/package-info.java similarity index 88% rename from it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datagenerator/package-info.java rename to it/common/src/main/java/org/apache/beam/it/common/dataflow/package-info.java index d563025ebeb4..eb10897523bc 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datagenerator/package-info.java +++ b/it/common/src/main/java/org/apache/beam/it/common/dataflow/package-info.java @@ -16,5 +16,5 @@ * limitations under the License. */ -/** Data generator for load tests. */ -package org.apache.beam.it.gcp.datagenerator; +/** Package for managing Dataflow jobs from integration tests. */ +package org.apache.beam.it.common.dataflow; diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/monitoring/MonitoringClient.java b/it/common/src/main/java/org/apache/beam/it/common/monitoring/MonitoringClient.java similarity index 99% rename from it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/monitoring/MonitoringClient.java rename to it/common/src/main/java/org/apache/beam/it/common/monitoring/MonitoringClient.java index 66550b5a5073..6b447d84512d 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/monitoring/MonitoringClient.java +++ b/it/common/src/main/java/org/apache/beam/it/common/monitoring/MonitoringClient.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.monitoring; +package org.apache.beam.it.common.monitoring; import com.google.api.gax.core.CredentialsProvider; import com.google.cloud.monitoring.v3.MetricServiceClient; diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/monitoring/package-info.java b/it/common/src/main/java/org/apache/beam/it/common/monitoring/package-info.java similarity index 94% rename from it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/monitoring/package-info.java rename to it/common/src/main/java/org/apache/beam/it/common/monitoring/package-info.java index 9cd79ea9cf12..731bbc3a6b25 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/monitoring/package-info.java +++ b/it/common/src/main/java/org/apache/beam/it/common/monitoring/package-info.java @@ -17,4 +17,4 @@ */ /** Package for querying metrics from cloud monitoring. */ -package org.apache.beam.it.gcp.monitoring; +package org.apache.beam.it.common.monitoring; 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 730ca23ee54b..85f2c8b0b018 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.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 efcea3aba267..1c5b3b3ce21e 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/main/resources/test-artifact.json b/it/common/src/main/resources/test-artifact.json similarity index 100% rename from it/google-cloud-platform/src/main/resources/test-artifact.json rename to it/common/src/main/resources/test-artifact.json 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 59b7931b603a..470741b399ad 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.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 491d32d0ce6a..137c35aaf4bb 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.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/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerTest.java b/it/common/src/test/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerTest.java similarity index 99% rename from it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerTest.java rename to it/common/src/test/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerTest.java index 53987dbe2670..62aac93b4eed 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerTest.java +++ b/it/common/src/test/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerTest.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.it.gcp.bigquery; +package org.apache.beam.it.common.bigquery; import static com.google.common.truth.Truth.assertThat; import static org.junit.Assert.assertThrows; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerUtilsTest.java b/it/common/src/test/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerUtilsTest.java similarity index 93% rename from it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerUtilsTest.java rename to it/common/src/test/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerUtilsTest.java index 432b7ef7362f..7fc848b45292 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryResourceManagerUtilsTest.java +++ b/it/common/src/test/java/org/apache/beam/it/common/bigquery/BigQueryResourceManagerUtilsTest.java @@ -15,9 +15,9 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.it.gcp.bigquery; +package org.apache.beam.it.common.bigquery; -import static org.apache.beam.it.gcp.bigquery.BigQueryResourceManagerUtils.checkValidTableId; +import static org.apache.beam.it.common.bigquery.BigQueryResourceManagerUtils.checkValidTableId; import static org.junit.Assert.assertThrows; import java.util.Arrays; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/AbstractPipelineLauncherTest.java b/it/common/src/test/java/org/apache/beam/it/common/dataflow/AbstractPipelineLauncherTest.java similarity index 99% rename from it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/AbstractPipelineLauncherTest.java rename to it/common/src/test/java/org/apache/beam/it/common/dataflow/AbstractPipelineLauncherTest.java index f2c33d70116b..72d5514a6e3c 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/AbstractPipelineLauncherTest.java +++ b/it/common/src/test/java/org/apache/beam/it/common/dataflow/AbstractPipelineLauncherTest.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.it.gcp.dataflow; +package org.apache.beam.it.common.dataflow; import static com.google.common.truth.Truth.assertThat; import static org.junit.Assert.assertThrows; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/DefaultPipelineLauncherTest.java b/it/common/src/test/java/org/apache/beam/it/common/dataflow/DefaultPipelineLauncherTest.java similarity index 91% rename from it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/DefaultPipelineLauncherTest.java rename to it/common/src/test/java/org/apache/beam/it/common/dataflow/DefaultPipelineLauncherTest.java index b6c0f8cdc58f..35510a0f1d1c 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/DefaultPipelineLauncherTest.java +++ b/it/common/src/test/java/org/apache/beam/it/common/dataflow/DefaultPipelineLauncherTest.java @@ -15,16 +15,16 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.it.gcp.dataflow; +package org.apache.beam.it.common.dataflow; +import static org.apache.beam.it.common.dataflow.DefaultPipelineLauncher.BEAM_METRICS_NAMESPACE; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import java.io.IOException; import java.time.Instant; import org.apache.beam.it.common.PipelineLauncher; -import org.apache.beam.it.gcp.IOLoadTestBase; -import org.apache.beam.it.gcp.IOLoadTestBase.PipelineMetricsType; +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; @@ -48,7 +48,7 @@ public void testPipelineMetrics() throws IOException { pipeline .apply(GenerateSequence.from(0).to(numElements)) - .apply(ParDo.of(new TimeMonitor<>(IOLoadTestBase.BEAM_METRICS_NAMESPACE, timeMetrics))) + .apply(ParDo.of(new TimeMonitor<>(BEAM_METRICS_NAMESPACE, timeMetrics))) .apply(ParDo.of(new IOLoadTestBase.CountingFn<>(counterMetrics))); PipelineLauncher.LaunchConfig options = diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/IOLoadTestBase.java b/it/common/src/test/java/org/apache/beam/it/common/dataflow/IOLoadTestBase.java similarity index 87% rename from it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/IOLoadTestBase.java rename to it/common/src/test/java/org/apache/beam/it/common/dataflow/IOLoadTestBase.java index 14770a429731..c35f402bd8ac 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/IOLoadTestBase.java +++ b/it/common/src/test/java/org/apache/beam/it/common/dataflow/IOLoadTestBase.java @@ -15,7 +15,9 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.it.gcp; +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; @@ -25,7 +27,7 @@ import java.util.UUID; import org.apache.beam.it.common.PipelineLauncher; import org.apache.beam.it.common.TestProperties; -import org.apache.beam.it.gcp.dataflow.DefaultPipelineLauncher; +import org.apache.beam.it.common.dataflow.DefaultPipelineLauncher.PipelineMetricsType; import org.apache.beam.sdk.metrics.Counter; import org.apache.beam.sdk.metrics.Metrics; import org.apache.beam.sdk.testutils.NamedTestResult; @@ -90,21 +92,6 @@ public void processElement(ProcessContext ctx) { } } - // To make PipelineLauncher.getMetric work in a unified way for both runner provided metrics and - // pipeline defined - // metrics, here we wrap Beam provided metrics as a pre-defined metrics name - // [name_space:metric_type:metric_name - // which will be recognized by getMetric method - public enum PipelineMetricsType { - COUNTER, - STARTTIME, - ENDTIME, - RUNTIME, - } - - /** Namespace for Beam provided pipeline metrics (set up by Metrics transform). */ - public static final String BEAM_METRICS_NAMESPACE = "BEAM_METRICS"; - /** Given a metrics name, return Beam metrics name. */ public static String getBeamMetricsName(PipelineMetricsType metricstype, String metricsName) { return BEAM_METRICS_NAMESPACE + ":" + metricstype + ":" + metricsName; diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/IOStressTestBase.java b/it/common/src/test/java/org/apache/beam/it/common/dataflow/IOStressTestBase.java similarity index 99% rename from it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/IOStressTestBase.java rename to it/common/src/test/java/org/apache/beam/it/common/dataflow/IOStressTestBase.java index 5c2fb74cd2fb..e08eec031eb5 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/IOStressTestBase.java +++ b/it/common/src/test/java/org/apache/beam/it/common/dataflow/IOStressTestBase.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.it.gcp; +package org.apache.beam.it.common.dataflow; import java.io.Serializable; import java.time.Duration; diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/LoadTestBase.java b/it/common/src/test/java/org/apache/beam/it/common/dataflow/LoadTestBase.java similarity index 98% rename from it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/LoadTestBase.java rename to it/common/src/test/java/org/apache/beam/it/common/dataflow/LoadTestBase.java index cd9ef52ed835..cd1b71dc7552 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/LoadTestBase.java +++ b/it/common/src/test/java/org/apache/beam/it/common/dataflow/LoadTestBase.java @@ -15,10 +15,10 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.it.gcp; +package org.apache.beam.it.common.dataflow; +import static org.apache.beam.it.common.dataflow.AbstractPipelineLauncher.RUNNER_V2; import static org.apache.beam.it.common.logging.LogStrings.formatForLogging; -import static org.apache.beam.it.gcp.dataflow.AbstractPipelineLauncher.RUNNER_V2; import com.google.api.gax.core.CredentialsProvider; import com.google.api.gax.core.FixedCredentialsProvider; @@ -44,8 +44,8 @@ import org.apache.beam.it.common.PipelineLauncher.LaunchInfo; import org.apache.beam.it.common.PipelineOperator; import org.apache.beam.it.common.TestProperties; -import org.apache.beam.it.gcp.bigquery.BigQueryResourceManager; -import org.apache.beam.it.gcp.monitoring.MonitoringClient; +import org.apache.beam.it.common.bigquery.BigQueryResourceManager; +import org.apache.beam.it.common.monitoring.MonitoringClient; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects; import org.checkerframework.checker.nullness.qual.Nullable; import org.junit.After; 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 0153573feaed..70dc6dcbc9c4 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.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.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 void testCreateArtifactInRunDir() { 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 void testUploadArtifact() throws IOException { 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 void testListArtifactsInMethodDirSinglePage() { 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 void testListArtifactsInMethodDirMultiplePages() { 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 3a46f2b94d83..164a75c06ba0 100644 --- a/it/google-cloud-platform/build.gradle +++ b/it/google-cloud-platform/build.gradle @@ -27,49 +27,33 @@ applyJavaNature( description = "Apache Beam :: IT :: Google Cloud Platform" ext.summary = "Integration test utilities for Google Cloud Platform." +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' @@ -84,6 +68,7 @@ dependencies { implementation 'com.google.cloud:google-cloud-secretmanager' provided 'com.google.api.grpc:proto-google-cloud-secretmanager-v1' + testImplementation project(path: ":it:common", configuration: "testRuntimeMigration") testImplementation project(path: ":sdks:java:io:google-cloud-platform") testImplementation project(path: ":sdks:java:extensions:protobuf", configuration: "testRuntimeMigration") testImplementation project(path: ":sdks:java:io:synthetic") 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 9759071b397e..000000000000 --- 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 28626f7edd18..da0cf5e55dbe 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 @@ import com.google.auto.value.AutoValue; import com.google.cloud.bigquery.TableId; +import org.apache.beam.it.common.bigquery.BigQueryResourceManager; import org.apache.beam.it.conditions.ConditionCheck; -import org.apache.beam.it.gcp.bigquery.BigQueryResourceManager; 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 00aa4003817f..7908246dce13 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,6 +30,7 @@ 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 0a9731b13c32..496c4b0e980e 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,6 +32,7 @@ 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 5a101e08d375..1778d70cf60a 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 b561c0c71fb6..16cf2b8f4d31 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 @@ -29,6 +29,7 @@ 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 f123ffa5954c..9ac03db1c741 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 @@ -43,8 +43,10 @@ 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.dataflow.DefaultPipelineLauncher.PipelineMetricsType; +import org.apache.beam.it.common.dataflow.IOLoadTestBase; import org.apache.beam.it.common.utils.ResourceManagerUtils; -import org.apache.beam.it.gcp.IOLoadTestBase; import org.apache.beam.sdk.io.Read; import org.apache.beam.sdk.io.gcp.bigquery.AvroWriteRequest; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryIOST.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryIOST.java index 14c83e48308a..495a4d4485d8 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 @@ -41,8 +41,10 @@ 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.dataflow.DefaultPipelineLauncher.PipelineMetricsType; +import org.apache.beam.it.common.dataflow.IOStressTestBase; import org.apache.beam.it.common.utils.ResourceManagerUtils; -import org.apache.beam.it.gcp.IOStressTestBase; import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions; import org.apache.beam.sdk.io.Read; import org.apache.beam.sdk.io.gcp.bigquery.AvroWriteRequest; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryStreamingLT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryStreamingLT.java index 6e511bd8e5c6..75a7be50a508 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryStreamingLT.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryStreamingLT.java @@ -41,7 +41,7 @@ import org.apache.beam.it.common.PipelineLauncher; import org.apache.beam.it.common.PipelineOperator; import org.apache.beam.it.common.TestProperties; -import org.apache.beam.it.gcp.IOLoadTestBase; +import org.apache.beam.it.common.dataflow.IOLoadTestBase; import org.apache.beam.runners.dataflow.DataflowRunner; import org.apache.beam.sdk.io.GenerateSequence; import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigtable/BigTableIOLT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigtable/BigTableIOLT.java index 410c992fe2d6..27773121e79d 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 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.utils.ResourceManagerUtils; -import org.apache.beam.it.gcp.IOLoadTestBase; import org.apache.beam.sdk.io.GenerateSequence; import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO; import org.apache.beam.sdk.testing.TestPipeline; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigtable/BigTableIOST.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigtable/BigTableIOST.java index 052a0839adf5..b86f13a0de10 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 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.it.common.utils.ResourceManagerUtils; -import org.apache.beam.it.gcp.IOStressTestBase; import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions; import org.apache.beam.sdk.io.Read; import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/ClassicTemplateClientTest.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/ClassicTemplateClientTest.java index 88c35589f2be..a50c0951005f 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/ClassicTemplateClientTest.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/ClassicTemplateClientTest.java @@ -18,10 +18,10 @@ package org.apache.beam.it.gcp.dataflow; import static com.google.common.truth.Truth.assertThat; -import static org.apache.beam.it.gcp.dataflow.AbstractPipelineLauncher.LEGACY_RUNNER; -import static org.apache.beam.it.gcp.dataflow.AbstractPipelineLauncher.PARAM_JOB_ID; -import static org.apache.beam.it.gcp.dataflow.AbstractPipelineLauncher.PARAM_JOB_TYPE; -import static org.apache.beam.it.gcp.dataflow.AbstractPipelineLauncher.PARAM_RUNNER; +import static org.apache.beam.it.common.dataflow.AbstractPipelineLauncher.LEGACY_RUNNER; +import static org.apache.beam.it.common.dataflow.AbstractPipelineLauncher.PARAM_JOB_ID; +import static org.apache.beam.it.common.dataflow.AbstractPipelineLauncher.PARAM_JOB_TYPE; +import static org.apache.beam.it.common.dataflow.AbstractPipelineLauncher.PARAM_RUNNER; import static org.junit.Assert.assertThrows; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; @@ -49,6 +49,7 @@ import org.apache.beam.it.common.PipelineLauncher.JobState; import org.apache.beam.it.common.PipelineLauncher.LaunchConfig; import org.apache.beam.it.common.PipelineLauncher.LaunchInfo; +import org.apache.beam.it.common.dataflow.AbstractPipelineLauncher; import org.junit.Rule; import org.junit.Test; import org.junit.runner.RunWith; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/FlexTemplateClientTest.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/FlexTemplateClientTest.java index 06f44437414a..b94d52e45afb 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/FlexTemplateClientTest.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/dataflow/FlexTemplateClientTest.java @@ -18,10 +18,10 @@ package org.apache.beam.it.gcp.dataflow; import static com.google.common.truth.Truth.assertThat; -import static org.apache.beam.it.gcp.dataflow.AbstractPipelineLauncher.LEGACY_RUNNER; -import static org.apache.beam.it.gcp.dataflow.AbstractPipelineLauncher.PARAM_JOB_ID; -import static org.apache.beam.it.gcp.dataflow.AbstractPipelineLauncher.PARAM_JOB_TYPE; -import static org.apache.beam.it.gcp.dataflow.AbstractPipelineLauncher.PARAM_RUNNER; +import static org.apache.beam.it.common.dataflow.AbstractPipelineLauncher.LEGACY_RUNNER; +import static org.apache.beam.it.common.dataflow.AbstractPipelineLauncher.PARAM_JOB_ID; +import static org.apache.beam.it.common.dataflow.AbstractPipelineLauncher.PARAM_JOB_TYPE; +import static org.apache.beam.it.common.dataflow.AbstractPipelineLauncher.PARAM_RUNNER; import static org.junit.Assert.assertThrows; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datagenerator/DataGenerator.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/datagenerator/DataGenerator.java similarity index 99% rename from it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datagenerator/DataGenerator.java rename to it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/datagenerator/DataGenerator.java index 99016b5dd3a4..0b838bddeb6d 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/datagenerator/DataGenerator.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/datagenerator/DataGenerator.java @@ -17,7 +17,7 @@ */ package org.apache.beam.it.gcp.datagenerator; -import static org.apache.beam.it.gcp.LoadTestBase.createConfig; +import static org.apache.beam.it.common.dataflow.LoadTestBase.createConfig; import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatPipeline; import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatResult; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/pubsub/PubSubIOST.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/pubsub/PubSubIOST.java index cb96db40b749..47058c75e399 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 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.it.common.utils.ResourceManagerUtils; -import org.apache.beam.it.gcp.IOStressTestBase; import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions; import org.apache.beam.sdk.extensions.protobuf.Proto3SchemaMessages.Primitive; import org.apache.beam.sdk.io.Read; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/pubsub/PubsubIOLT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/pubsub/PubsubIOLT.java index 477c79dd3bc7..da43fca2eaa3 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 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.utils.ResourceManagerUtils; -import org.apache.beam.it.gcp.IOLoadTestBase; import org.apache.beam.runners.direct.DirectOptions; import org.apache.beam.sdk.extensions.protobuf.Proto3SchemaMessages.Primitive; import org.apache.beam.sdk.io.Read; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/spanner/SpannerIOLT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/spanner/SpannerIOLT.java index 8f3de1b58f40..4441b8dcd6b6 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 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.utils.ResourceManagerUtils; -import org.apache.beam.it.gcp.IOLoadTestBase; import org.apache.beam.sdk.io.GenerateSequence; import org.apache.beam.sdk.io.gcp.spanner.SpannerIO; import org.apache.beam.sdk.io.synthetic.SyntheticSourceOptions; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/spanner/SpannerIOST.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/spanner/SpannerIOST.java index 76312b3e10fb..4eb7030c23c3 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 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.it.common.utils.ResourceManagerUtils; -import org.apache.beam.it.gcp.IOStressTestBase; import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions; import org.apache.beam.sdk.io.Read; import org.apache.beam.sdk.io.gcp.spanner.SpannerIO; diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/FileBasedIOLT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/FileBasedIOLT.java index 58ba88d900e8..ac1a7fc103c6 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 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.it.gcp.IOLoadTestBase; import org.apache.beam.sdk.io.Compression; import org.apache.beam.sdk.io.Read; import org.apache.beam.sdk.io.TextIO; diff --git a/it/iceberg/build.gradle b/it/iceberg/build.gradle new file mode 100644 index 000000000000..f11374a1ca4f --- /dev/null +++ b/it/iceberg/build.gradle @@ -0,0 +1,134 @@ +/* + * 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. + */ + +import org.apache.beam.gradle.IoPerformanceTestUtilities + +plugins { id 'org.apache.beam.module' } +applyJavaNature( + automaticModuleName: 'org.apache.beam.it.iceberg', + exportJavadoc: false, + // iceberg ended support for Java 8 in 1.7.0 + requireJavaVersion: JavaVersion.VERSION_11, +) + +description = "Apache Beam :: IT :: Iceberg" +ext.summary = "Integration test utilities for Iceberg." + +def iceberg_version = "1.10.0" +def parquet_version = "1.16.0" +def hive_version = "3.1.3" + +def hadoopVersions = [ + "336": "3.3.6", +] + +hadoopVersions.each {kv -> configurations.create("hadoopVersion$kv.key")} + +evaluationDependsOn(":it:common") + +dependencies { + provided "org.immutables:value:2.8.8" + permitUnusedDeclared "org.immutables:value:2.8.8" + 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" + runtimeOnly "org.apache.iceberg:iceberg-azure:$iceberg_version" + runtimeOnly "org.apache.iceberg:iceberg-azure-bundle:$iceberg_version" + runtimeOnly library.java.bigdataoss_gcs_connector + runtimeOnly library.java.bigdataoss_util_hadoop + runtimeOnly library.java.hadoop_client + + testImplementation project(":sdks:java:managed") + testImplementation library.java.bigdataoss_gcsio + testImplementation library.java.bigdataoss_util_hadoop + testImplementation "org.apache.parquet:parquet-avro:$parquet_version" + testImplementation "org.apache.iceberg:iceberg-data:$iceberg_version" + testImplementation project(path: ":sdks:java:core", configuration: "shadowTest") + testImplementation library.java.junit + testImplementation library.java.hamcrest + testImplementation project(path: ":sdks:java:extensions:avro") + testImplementation 'org.awaitility:awaitility:4.2.0' + + // Hive catalog test dependencies + testImplementation project(path: ":sdks:java:io:iceberg:hive") + testImplementation "org.apache.iceberg:iceberg-common:$iceberg_version" + testImplementation ("org.apache.iceberg:iceberg-hive-metastore:$iceberg_version") + testImplementation ("org.apache.hive:hive-metastore:$hive_version") + testImplementation "org.assertj:assertj-core:3.11.1" + testRuntimeOnly ("org.apache.hive.hcatalog:hive-hcatalog-core:$hive_version") { + exclude group: "org.apache.hive", module: "hive-exec" + exclude group: "org.apache.parquet", module: "parquet-hadoop-bundle" + } + + // BigQueryMetastore catalog dep + testImplementation project(path: ":sdks:java:io:iceberg:bqms", configuration: "shadow") + testImplementation project(":sdks:java:io:google-cloud-platform") + testImplementation library.java.google_api_services_bigquery + testImplementation 'com.google.cloud:google-cloud-storage' + + testImplementation library.java.google_auth_library_oauth2_http + testRuntimeOnly library.java.slf4j_jdk14 + testImplementation project(path: ":runners:direct-java", configuration: "shadow") + testRuntimeOnly project(path: ":runners:google-cloud-dataflow-java") + testRuntimeOnly project(path: ":sdks:java:harness") + hadoopVersions.each {kv -> + "hadoopVersion$kv.key" "org.apache.hadoop:hadoop-client:$kv.value" + "hadoopVersion$kv.key" "org.apache.hadoop:hadoop-minicluster:$kv.value" + "hadoopVersion$kv.key" "org.apache.hadoop:hadoop-hdfs-client:$kv.value" + "hadoopVersion$kv.key" "org.apache.hadoop:hadoop-mapreduce-client-core:$kv.value" + } + + 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") + testRuntimeOnly library.java.slf4j_simple +} + +configurations.all { + // iceberg-core needs avro:1.12.0 + resolutionStrategy.force 'org.apache.avro:avro:1.12.0' + // TODO(https://github.com/apache/beam/issues/38515): + // Remove below pins when parquet-hadoop upgrades to hadoop-common:3.4.2 + resolutionStrategy.force 'org.apache.hadoop:hadoop-common:3.3.6' + resolutionStrategy.force 'org.apache.hadoop:hadoop-client:3.3.6' + resolutionStrategy.force 'org.apache.hadoop:hadoop-hdfs:3.3.6' + resolutionStrategy.force 'org.apache.hadoop:hadoop-hdfs-client:3.3.6' +} + +hadoopVersions.each {kv -> + configurations."hadoopVersion$kv.key" { + resolutionStrategy { + force "org.apache.hadoop:hadoop-client:$kv.value" + force "org.apache.hadoop:hadoop-common:$kv.value" + force "org.apache.hadoop:hadoop-mapreduce-client-core:$kv.value" + force "org.apache.hadoop:hadoop-minicluster:$kv.value" + force "org.apache.hadoop:hadoop-hdfs:$kv.value" + force "org.apache.hadoop:hadoop-hdfs-client:$kv.value" + } + } +} + +tasks.register( + "IcebergPerformanceTest", IoPerformanceTestUtilities.IoPerformanceTest, project, 'iceberg', 'IcebergIOLT', + ['configuration':'large','project':'apache-beam-testing', 'artifactBucket':'io-performance-temp'] + + System.properties +) 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 new file mode 100644 index 000000000000..2a6098a1518e --- /dev/null +++ b/it/iceberg/src/test/java/org/apache/beam/it/iceberg/IcebergIOLT.java @@ -0,0 +1,363 @@ +/* + * 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. + * + *

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. + * + *

Example trigger command: + * + * 

+ * ./gradlew :it:iceberg:integrationTest \
+ *   --tests "org.apache.beam.it.iceberg.IcebergIOLT" \
+ *   -Dconfiguration=local \
+ *   -Dproject=[gcpProject] \
+ *   -DartifactBucket=[artifactBucket]
+ *
+ */ +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 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 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 { + + private final int valueSizeBytes; + private transient Random random; + + private MapToIcebergFormat(int valueSizeBytes) { + if (valueSizeBytes <= 0) { + throw new IllegalArgumentException("valueSizeBytes must be positive."); + } + + this.valueSizeBytes = valueSizeBytes; + } + + @Setup + public void setup() { + random = new Random(); + } + + @ProcessElement + public void processElement(@Element Long index, OutputReceiver 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. + */ + random.setSeed(index); + random.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); + + abstract Builder setValueSizeBytes(int valueSizeBytes); + + abstract Configuration build(); + } + } +} diff --git a/it/kafka/build.gradle b/it/kafka/build.gradle index 0496c423bebd..32ef0623fea9 100644 --- a/it/kafka/build.gradle +++ b/it/kafka/build.gradle @@ -42,8 +42,7 @@ dependencies { permitUsedUndeclared library.java.guava testImplementation library.java.mockito_core testRuntimeOnly library.java.slf4j_simple - // TODO: Remove the below dependencies after KafkaIOLT has been moved to the right directory. - testImplementation project(path: ":it:google-cloud-platform") + testImplementation project(path: ":it:common", configuration: "testRuntimeMigration") testImplementation project(path: ":sdks:java:io:kafka") testImplementation project(path: ":sdks:java:io:synthetic") testImplementation project(path: ":runners:direct-java") diff --git a/it/kafka/src/test/java/org/apache/beam/it/kafka/KafkaIOLT.java b/it/kafka/src/test/java/org/apache/beam/it/kafka/KafkaIOLT.java index c20699d61a55..1607ca1cd674 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 @@ -30,8 +30,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.dataflow.DefaultPipelineLauncher.PipelineMetricsType; +import org.apache.beam.it.common.dataflow.IOLoadTestBase; import org.apache.beam.it.common.utils.ResourceManagerUtils; -import org.apache.beam.it.gcp.IOLoadTestBase; import org.apache.beam.runners.direct.DirectOptions; import org.apache.beam.sdk.io.kafka.KafkaIO; import org.apache.beam.sdk.io.synthetic.SyntheticOptions; diff --git a/it/kafka/src/test/java/org/apache/beam/it/kafka/KafkaIOST.java b/it/kafka/src/test/java/org/apache/beam/it/kafka/KafkaIOST.java index bb1cb37a11b6..d6231a69f1e7 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 @@ -34,7 +34,8 @@ import org.apache.beam.it.common.PipelineLauncher; import org.apache.beam.it.common.PipelineOperator; import org.apache.beam.it.common.TestProperties; -import org.apache.beam.it.gcp.IOStressTestBase; +import org.apache.beam.it.common.dataflow.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; import org.apache.beam.sdk.io.kafka.KafkaIO; diff --git a/it/truthmatchers/build.gradle b/it/truthmatchers/build.gradle index fd54ad591a0a..2e21c1b5df04 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 3b1251307d0d..92407e5a2fc9 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 artifacts) { - return assertAbout(ArtifactsSubject.records()).that(artifacts); + return Truth.assertAbout(ArtifactsSubject.records()).that(artifacts); } /** @@ -47,7 +44,7 @@ public static ArtifactsSubject assertThatArtifacts(List artifacts) { * @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 static ArtifactsSubject assertThatArtifact(Artifact artifact) { * @return Truth Subject to chain assertions. */ public static RecordsSubject assertThatGenericRecords(List 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 ab5b49699484..e2518280345d 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.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 void hasContent(String content) { */ 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 RecordsSubject asJsonRecords() { throw new RuntimeException("Error reading " + artifact.name() + " as JSON.", e); } } - return assertThatRecords(allRecords); + return PipelineAsserts.assertThatRecords(allRecords); } } diff --git a/settings.gradle.kts b/settings.gradle.kts index d9dbbf9021ef..5ee8f8a4d394 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -122,6 +122,7 @@ include(":it:conditions") include(":it:datadog") include(":it:elasticsearch") include(":it:google-cloud-platform") +include(":it:iceberg") include(":it:jdbc") include(":it:kafka") include(":it:testcontainers") diff --git a/website/www/site/content/en/performance/_index.md b/website/www/site/content/en/performance/_index.md index a0eaba2aa0ec..0181965fc105 100644 --- a/website/www/site/content/en/performance/_index.md +++ b/website/www/site/content/en/performance/_index.md @@ -38,6 +38,7 @@ writing to various Beam IOs. - [BigQuery](/performance/bigquery) - [BigTable](/performance/bigtable) - [TextIO](/performance/textio) +- [IcebergIO](/performance/icebergio) # Measured Beam Python ML Pipelines diff --git a/website/www/site/content/en/performance/icebergio/_index.md b/website/www/site/content/en/performance/icebergio/_index.md new file mode 100644 index 000000000000..e618865097b4 --- /dev/null +++ b/website/www/site/content/en/performance/icebergio/_index.md @@ -0,0 +1,50 @@ +--- +title: "IcebergIO Performance" +--- + + + +# IcebergIO Performance + +The following graphs show various metrics when reading from or writing to Apache Iceberg tables using +IcebergIO. See the [glossary](/performance/glossary) for definitions. + +## Read + +### What is the estimated cost of reading from Apache Iceberg tables using IcebergIO? + +{{< performance_looks io="icebergio" read_or_write="read" section="test_name" >}} + +### How has various metrics changed when reading from Apache Iceberg tables using IcebergIO for different Beam SDK versions? + +{{< performance_looks io="icebergio" read_or_write="read" section="version" >}} + +### How has various metrics changed over time when reading from Apache Iceberg tables using IcebergIO? + +{{< performance_looks io="icebergio" read_or_write="read" section="date" >}} + +## Write + +### What is the estimated cost of writing to Apache Iceberg tables using IcebergIO? + +{{< performance_looks io="icebergio" read_or_write="write" section="test_name" >}} + +### How has various metrics changed when writing to Apache Iceberg tables using IcebergIO for different Beam SDK versions? + +{{< performance_looks io="icebergio" read_or_write="write" section="version" >}} + +### How has various metrics changed over time when writing to Apache Iceberg tables using IcebergIO? + +{{< performance_looks io="icebergio" read_or_write="write" section="date" >}} diff --git a/website/www/site/data/performance.yaml b/website/www/site/data/performance.yaml index 42fb5bee92f5..e996ea262911 100644 --- a/website/www/site/data/performance.yaml +++ b/website/www/site/data/performance.yaml @@ -107,6 +107,37 @@ looks: title: AvgInputThroughputBytesPerSec by Version - id: fVVHhXCrHNgBG52TJsTjR8VbmWCCQnVN title: AvgInputThroughputElementsPerSec by Version + icebergio: + read: + folder: 113 + test_name: + - id: x5R7w8VnzkccgzrBTWQ82J7JBNtkGxH2 + title: Read IcebergIO RunTime and EstimatedCost + date: + - id: HQsKkmn7cS7m5bCYzPBYqnnFM6MXvJgV + title: AvgOutputThroughputBytesPerSec by Date + - id: MNYqD2ZCwV2XfYjwDg4NDVT7vr4Ytpz2 + title: AvgOutputThroughputElementsPerSec by Date + version: + - id: Cf7wGQfrM4hn8Wdf64ZjwshrHHdZKmKs + title: AvgOutputThroughputBytesPerSec by Version + - id: MGSX85MhCnCFFWYzSCGS36JRWxcBccqd + title: AvgOutputThroughputElementsPerSec by Version + write: + folder: 114 + test_name: + - id: yvvWB2JbbbFQ4R5MTwF2hbVW2X4bDxnY + title: Write IcebergIO RunTime and EstimatedCost + date: + - id: Gcw6jfhvvkkHKDwqqTkhZtYYNNH2Z8DZ + title: AvgInputThroughputBytesPerSec by Date + - id: RYtRGKsxBRkyRgWk6RGjshFnZPk76vhM + title: AvgInputThroughputElementsPerSec by Date + version: + - id: B5q25ktnXNhHxpYrGmqzgcGMgX26zvM3 + title: AvgInputThroughputBytesPerSec by Version + - id: VfMWk6rk26JQNr8wRpsK5Gk6XQPfd7vW + title: AvgInputThroughputElementsPerSec by Version pytorchbertbase: write: folder: 76