From 0519b2f50959e80cc0f4e11b5480fb3ecaa2356d Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Mon, 3 Aug 2026 21:01:44 +0000 Subject: [PATCH 1/3] add delta lake to iceberg IT --- .../yaml/DeltaLakeToIcebergYaml.java | 164 ++++++++++++++++ .../python/options/deltalake_options.yaml | 18 ++ .../yaml/DeltaLakeToIcebergYamlIT.java | 180 ++++++++++++++++++ 3 files changed, 362 insertions(+) create mode 100644 yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java create mode 100644 yaml/src/main/python/options/deltalake_options.yaml create mode 100644 yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java new file mode 100644 index 0000000000..1649c5a6e5 --- /dev/null +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java @@ -0,0 +1,164 @@ +/* + * Copyright (C) 2026 Google LLC + * + * Licensed 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 com.google.cloud.teleport.templates.yaml; + +import com.google.cloud.teleport.metadata.Template; +import com.google.cloud.teleport.metadata.TemplateCategory; +import com.google.cloud.teleport.metadata.TemplateParameter; +import org.apache.beam.sdk.options.Validation; + +@Template( + name = "DeltaLake_To_Iceberg_Yaml", + category = TemplateCategory.BATCH, + type = Template.TemplateType.YAML, + displayName = "Delta Lake to Iceberg (YAML)", + description = "The Delta Lake to Iceberg template is a batch pipeline that reads data from a Delta Lake table and outputs the records to an Apache Iceberg table.", + flexContainerName = "pipeline-yaml", + yamlTemplateFile = "DeltaLakeToIceberg.yaml", + filesToCopy = {"main.py", "requirements.txt", "options/deltalake_options.yaml", "options/iceberg_options.yaml"}, + documentation = "", + contactInformation = "https://cloud.google.com/support", + requirements = {"The Input Delta Lake table must exist and be accessible.", + "The Output Iceberg table must exist or be created, and the warehouse must be accessible." + }, + streaming = false, + hidden = false) +public interface DeltaLakeToIcebergYaml { + + @TemplateParameter.Text( + order = 1, + name = "deltaLakeTable", + optional = false, + description = "A GCS path to the Delta Lake table.", + helpText = "The GCS path to the Delta Lake table, e.g., gs://your-bucket/path/to/table.", + example = "gs://your-bucket/path/to/table" + ) + @Validation.Required + String getDeltaLakeTable(); + + @TemplateParameter.Text( + order = 2, + name = "deltaLakeHadoopConfig", + optional = true, + description = "Properties passed to Hadoop Configuration.", + helpText = "A map of properties to pass to Hadoop Configuration, e.g. key-value pairs.", + example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}" + ) + String getDeltaLakeHadoopConfig(); + + @TemplateParameter.Text( + order = 3, + name = "table", + optional = false, + description = "A fully-qualified table identifier.", + helpText = "A fully-qualified table identifier, e.g., my_dataset.my_table.", + example = "my_dataset.my_table" + ) + @Validation.Required + String getTable(); + + @TemplateParameter.Text( + order = 4, + name = "catalogName", + optional = false, + description = "Name of the catalog containing the table.", + helpText = "The name of the Iceberg catalog that contains the table.", + example = "my_hadoop_catalog" + ) + @Validation.Required + String getCatalogName(); + + @TemplateParameter.Text( + order = 5, + name = "catalogProperties", + optional = false, + description = "Properties used to set up the Iceberg catalog.", + helpText = "A map of properties for setting up the Iceberg catalog.", + example = "{\"type\": \"hadoop\", \"warehouse\": \"gs://your-bucket/warehouse\"}" + ) + @Validation.Required + String getCatalogProperties(); + + @TemplateParameter.Text( + order = 6, + name = "configProperties", + optional = true, + description = "Properties passed to the Hadoop Configuration.", + helpText = "A map of properties to pass to the Hadoop Configuration.", + example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}" + ) + String getConfigProperties(); + + @TemplateParameter.Text( + order = 7, + name = "drop", + optional = true, + description = "A list of field names to drop from the input record before writing.", + helpText = "A list of field names to drop. Mutually exclusive with 'keep' and 'only'.", + example = "[\"field_to_drop_1\", \"field_to_drop_2\"]" + ) + String getDrop(); + + @TemplateParameter.Text( + order = 8, + name = "filter", + optional = true, + description = "An optional filter expression to apply to the input records.", + helpText = "A filter expression to apply to records from the Iceberg table.", + example = "age > 18" + ) + String getFilter(); + + @TemplateParameter.Text( + order = 9, + name = "keep", + optional = true, + description = "A list of field names to keep in the input record.", + helpText = "A list of field names to keep. Mutually exclusive with 'drop' and 'only'.", + example = "[\"field_to_keep_1\", \"field_to_keep_2\"]" + ) + String getKeep(); + + @TemplateParameter.Text( + order = 10, + name = "only", + optional = true, + description = "The name of a single record field that should be written.", + helpText = "The name of a single field to write. Mutually exclusive with 'keep' and 'drop'.", + example = "my_record_field" + ) + String getOnly(); + + @TemplateParameter.Text( + order = 11, + name = "partitionFields", + optional = true, + description = "Fields used to create a partition spec for new tables.", + helpText = "A list of fields and transforms for partitioning, e.g., ['day(ts)', 'category'].", + example = "[\"day(ts)\", \"bucket(id, 4)\"]" + ) + String getPartitionFields(); + + @TemplateParameter.Text( + order = 12, + name = "tableProperties", + optional = true, + description = "Iceberg table properties to be set on table creation.", + helpText = "A map of Iceberg table properties to set when the table is created.", + example = "{\"commit.retry.num-retries\": \"2\"}" + ) + String getTableProperties(); +} diff --git a/yaml/src/main/python/options/deltalake_options.yaml b/yaml/src/main/python/options/deltalake_options.yaml new file mode 100644 index 0000000000..26940eb9b6 --- /dev/null +++ b/yaml/src/main/python/options/deltalake_options.yaml @@ -0,0 +1,18 @@ +options: + - name: "deltalake_read_options" + parameters: + - order: 1 + name: "deltaLakeTable" + description: "A GCS path to the Delta Lake table." + help: "The GCS path to the Delta Lake table, e.g., gs://your-bucket/path/to/table." + example: "gs://your-bucket/path/to/table" + required: true + type: text + - order: 2 + name: "deltaLakeHadoopConfig" + description: "Properties passed to Hadoop Configuration." + help: "A map of properties to pass to Hadoop Configuration, e.g. key-value pairs." + example: '{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"}' + required: false + type: map + diff --git a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java new file mode 100644 index 0000000000..684961ea4d --- /dev/null +++ b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java @@ -0,0 +1,180 @@ +/* + * Copyright (C) 2026 Google LLC + * + * Licensed 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 com.google.cloud.teleport.templates.yaml; + +import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatPipeline; +import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatResult; +import static org.junit.Assert.assertEquals; + +import com.google.cloud.teleport.it.iceberg.IcebergResourceManager; +import com.google.cloud.teleport.metadata.SkipDirectRunnerTest; +import com.google.cloud.teleport.metadata.TemplateIntegrationTest; +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import org.apache.beam.it.common.PipelineLauncher.LaunchConfig; +import org.apache.beam.it.common.PipelineLauncher.LaunchInfo; +import org.apache.beam.it.common.PipelineOperator; +import org.apache.beam.it.common.utils.ResourceManagerUtils; +import org.apache.beam.it.gcp.TemplateTestBase; +import org.apache.beam.it.gcp.artifacts.utils.ParquetTestUtil; +import org.apache.iceberg.Schema; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.types.Types; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Integration test for {@link DeltaLakeToIcebergYaml} template. */ +@Category({TemplateIntegrationTest.class, SkipDirectRunnerTest.class}) +@TemplateIntegrationTest(DeltaLakeToIcebergYaml.class) +@RunWith(JUnit4.class) +public class DeltaLakeToIcebergYamlIT extends TemplateTestBase { + + private IcebergResourceManager icebergResourceManager; + + private static final String CATALOG_NAME = "hadoop_catalog"; + private final String namespace = + "deltalake_iceberg_ns_" + UUID.randomUUID().toString().replace("-", ""); + private static final String ICEBERG_TABLE_NAME = "iceberg_table"; + private final String icebergTableIdentifier = namespace + "." + ICEBERG_TABLE_NAME; + + @Before + public void setUp() throws IOException { + gcsClient.registerTempDir(namespace); + + // Initialize Iceberg resource manager + icebergResourceManager = + IcebergResourceManager.builder(testName) + .setCatalogName(CATALOG_NAME) + .setCatalogProperties(getCatalogProperties()) + .build(); + } + + @After + public void tearDown() { + ResourceManagerUtils.cleanResources(icebergResourceManager); + } + + @Test + public void testDeltaLakeToIceberg() throws IOException { + // 1. Arrange: Create Delta Lake source table in GCS + String deltaTableDir = "delta-table"; + org.apache.avro.Schema avroSchema = + new org.apache.avro.Schema.Parser() + .parse( + "{\"type\":\"record\",\"name\":\"test_record\",\"fields\":[" + + "{\"name\":\"id\",\"type\":\"string\"}," + + "{\"name\":\"state\",\"type\":\"string\"}," + + "{\"name\":\"price\",\"type\":\"double\"}" + + "]}"); + org.apache.avro.generic.GenericRecord avroRecord = + new org.apache.avro.generic.GenericData.Record(avroSchema); + avroRecord.put("id", "007"); + avroRecord.put("state", "CA"); + avroRecord.put("price", 26.23); + byte[] parquetBytes = + ParquetTestUtil.createParquetFile(avroSchema, List.of(avroRecord)); + + // Upload data Parquet file + gcsClient.createArtifact(deltaTableDir + "/part-00000.parquet", parquetBytes); + + // Create and upload Delta Lake transaction log + String commitContent = + "{\"protocol\":{\"minReaderVersion\":1,\"minWriterVersion\":2}}\n" + + "{\"metaData\":{\"id\":\"test-id\",\"format\":{\"provider\":\"parquet\",\"options\":{}}," + + "\"schemaString\":\"{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[" + + "{\\\"name\\\":\\\"id\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"state\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"price\\\",\\\"type\\\":\\\"double\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}" + + "]}\",\"partitionColumns\":[],\"configuration\":{},\"createdAt\":123456789}}\n" + + "{\"add\":{\"path\":\"part-00000.parquet\",\"partitionValues\":{},\"size\":774," + + "\"modificationTime\":123456789,\"dataChange\":true}}"; + + gcsClient.createArtifact( + deltaTableDir + "/_delta_log/00000000000000000000.json", commitContent); + + String deltaTableGcsPath = getGcsPath(deltaTableDir); + + // 2. Arrange: Create destination Iceberg table + icebergResourceManager.createNamespace(namespace); + Schema icebergSchema = + new Schema( + Types.NestedField.required(1, "id", Types.StringType.get()), + Types.NestedField.required(2, "state", Types.StringType.get()), + Types.NestedField.required(3, "price", Types.DoubleType.get())); + icebergResourceManager.createTable(icebergTableIdentifier, icebergSchema); + + // 3. Act: Configure options and launch template + LaunchConfig.Builder options = + LaunchConfig.builder(testName, specPath) + .addParameter("deltaLakeTable", deltaTableGcsPath) + .addParameter( + "deltaLakeHadoopConfig", + new org.json.JSONObject(getGcsHadoopConfig()).toString()) + .addParameter("table", icebergTableIdentifier) + .addParameter("catalogName", CATALOG_NAME) + .addParameter( + "catalogProperties", new org.json.JSONObject(getCatalogProperties()).toString()); + + LaunchInfo info = launchTemplate(options); + assertThatPipeline(info).isRunning(); + + PipelineOperator.Result result = pipelineOperator().waitUntilDone(createConfig(info)); + + // 4. Assert + assertThatResult(result).isLaunchFinished(); + + List records = icebergResourceManager.read(icebergTableIdentifier); + assertEquals(1, records.size()); + + Record record = records.get(0); + assertEquals("007", record.getField("id")); + assertEquals("CA", record.getField("state")); + assertEquals(26.23, record.getField("price")); + } + + @Override + protected PipelineOperator.Config createConfig(LaunchInfo info) { + return PipelineOperator.Config.builder() + .setJobId(info.jobId()) + .setProject(PROJECT) + .setRegion(REGION) + .build(); + } + + private Map getCatalogProperties() { + return Map.of( + "type", "rest", + "uri", "https://biglake.googleapis.com/iceberg/v1beta/restcatalog", + "warehouse", "gs://" + gcsClient.getBucket(), + "header.x-goog-user-project", PROJECT, + "rest.auth.type", "org.apache.iceberg.gcp.auth.GoogleAuthManager", + "rest-metrics-reporting-enabled", "false"); + } + + private Map getGcsHadoopConfig() { + return Map.of( + "fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem", + "fs.AbstractFileSystem.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS", + "fs.gs.auth.type", "APPLICATION_DEFAULT", + "fs.gs.project.id", PROJECT); + } +} From b8e1c1a3a175ffdccf60f768667c487375142637 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Mon, 3 Aug 2026 21:14:38 +0000 Subject: [PATCH 2/3] fix spotless and rerun auto-generation --- .../yaml/DeltaLakeToIcebergYaml.java | 51 +++++++++---------- .../yaml/DeltaLakeToIcebergYamlIT.java | 6 +-- 2 files changed, 25 insertions(+), 32 deletions(-) diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java index 1649c5a6e5..6052128284 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java @@ -24,14 +24,21 @@ name = "DeltaLake_To_Iceberg_Yaml", category = TemplateCategory.BATCH, type = Template.TemplateType.YAML, - displayName = "Delta Lake to Iceberg (YAML)", - description = "The Delta Lake to Iceberg template is a batch pipeline that reads data from a Delta Lake table and outputs the records to an Apache Iceberg table.", + displayName = "Delta Lake to Lakehouse", + description = + "The Delta Lake to Iceberg template is a batch pipeline that reads data from a Delta Lake table and outputs the records to an Apache Iceberg table.", flexContainerName = "pipeline-yaml", yamlTemplateFile = "DeltaLakeToIceberg.yaml", - filesToCopy = {"main.py", "requirements.txt", "options/deltalake_options.yaml", "options/iceberg_options.yaml"}, + filesToCopy = { + "main.py", + "requirements.txt", + "options/deltalake_options.yaml", + "options/iceberg_options.yaml" + }, documentation = "", contactInformation = "https://cloud.google.com/support", - requirements = {"The Input Delta Lake table must exist and be accessible.", + requirements = { + "The Input Delta Lake table must exist and be accessible.", "The Output Iceberg table must exist or be created, and the warehouse must be accessible." }, streaming = false, @@ -44,8 +51,7 @@ public interface DeltaLakeToIcebergYaml { optional = false, description = "A GCS path to the Delta Lake table.", helpText = "The GCS path to the Delta Lake table, e.g., gs://your-bucket/path/to/table.", - example = "gs://your-bucket/path/to/table" - ) + example = "gs://your-bucket/path/to/table") @Validation.Required String getDeltaLakeTable(); @@ -55,8 +61,7 @@ public interface DeltaLakeToIcebergYaml { optional = true, description = "Properties passed to Hadoop Configuration.", helpText = "A map of properties to pass to Hadoop Configuration, e.g. key-value pairs.", - example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}" - ) + example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}") String getDeltaLakeHadoopConfig(); @TemplateParameter.Text( @@ -65,8 +70,7 @@ public interface DeltaLakeToIcebergYaml { optional = false, description = "A fully-qualified table identifier.", helpText = "A fully-qualified table identifier, e.g., my_dataset.my_table.", - example = "my_dataset.my_table" - ) + example = "my_dataset.my_table") @Validation.Required String getTable(); @@ -76,8 +80,7 @@ public interface DeltaLakeToIcebergYaml { optional = false, description = "Name of the catalog containing the table.", helpText = "The name of the Iceberg catalog that contains the table.", - example = "my_hadoop_catalog" - ) + example = "my_hadoop_catalog") @Validation.Required String getCatalogName(); @@ -87,8 +90,7 @@ public interface DeltaLakeToIcebergYaml { optional = false, description = "Properties used to set up the Iceberg catalog.", helpText = "A map of properties for setting up the Iceberg catalog.", - example = "{\"type\": \"hadoop\", \"warehouse\": \"gs://your-bucket/warehouse\"}" - ) + example = "{\"type\": \"hadoop\", \"warehouse\": \"gs://your-bucket/warehouse\"}") @Validation.Required String getCatalogProperties(); @@ -98,8 +100,7 @@ public interface DeltaLakeToIcebergYaml { optional = true, description = "Properties passed to the Hadoop Configuration.", helpText = "A map of properties to pass to the Hadoop Configuration.", - example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}" - ) + example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}") String getConfigProperties(); @TemplateParameter.Text( @@ -108,8 +109,7 @@ public interface DeltaLakeToIcebergYaml { optional = true, description = "A list of field names to drop from the input record before writing.", helpText = "A list of field names to drop. Mutually exclusive with 'keep' and 'only'.", - example = "[\"field_to_drop_1\", \"field_to_drop_2\"]" - ) + example = "[\"field_to_drop_1\", \"field_to_drop_2\"]") String getDrop(); @TemplateParameter.Text( @@ -118,8 +118,7 @@ public interface DeltaLakeToIcebergYaml { optional = true, description = "An optional filter expression to apply to the input records.", helpText = "A filter expression to apply to records from the Iceberg table.", - example = "age > 18" - ) + example = "age > 18") String getFilter(); @TemplateParameter.Text( @@ -128,8 +127,7 @@ public interface DeltaLakeToIcebergYaml { optional = true, description = "A list of field names to keep in the input record.", helpText = "A list of field names to keep. Mutually exclusive with 'drop' and 'only'.", - example = "[\"field_to_keep_1\", \"field_to_keep_2\"]" - ) + example = "[\"field_to_keep_1\", \"field_to_keep_2\"]") String getKeep(); @TemplateParameter.Text( @@ -138,8 +136,7 @@ public interface DeltaLakeToIcebergYaml { optional = true, description = "The name of a single record field that should be written.", helpText = "The name of a single field to write. Mutually exclusive with 'keep' and 'drop'.", - example = "my_record_field" - ) + example = "my_record_field") String getOnly(); @TemplateParameter.Text( @@ -148,8 +145,7 @@ public interface DeltaLakeToIcebergYaml { optional = true, description = "Fields used to create a partition spec for new tables.", helpText = "A list of fields and transforms for partitioning, e.g., ['day(ts)', 'category'].", - example = "[\"day(ts)\", \"bucket(id, 4)\"]" - ) + example = "[\"day(ts)\", \"bucket(id, 4)\"]") String getPartitionFields(); @TemplateParameter.Text( @@ -158,7 +154,6 @@ public interface DeltaLakeToIcebergYaml { optional = true, description = "Iceberg table properties to be set on table creation.", helpText = "A map of Iceberg table properties to set when the table is created.", - example = "{\"commit.retry.num-retries\": \"2\"}" - ) + example = "{\"commit.retry.num-retries\": \"2\"}") String getTableProperties(); } diff --git a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java index 684961ea4d..d7e8e80e2e 100644 --- a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java +++ b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java @@ -90,8 +90,7 @@ public void testDeltaLakeToIceberg() throws IOException { avroRecord.put("id", "007"); avroRecord.put("state", "CA"); avroRecord.put("price", 26.23); - byte[] parquetBytes = - ParquetTestUtil.createParquetFile(avroSchema, List.of(avroRecord)); + byte[] parquetBytes = ParquetTestUtil.createParquetFile(avroSchema, List.of(avroRecord)); // Upload data Parquet file gcsClient.createArtifact(deltaTableDir + "/part-00000.parquet", parquetBytes); @@ -127,8 +126,7 @@ public void testDeltaLakeToIceberg() throws IOException { LaunchConfig.builder(testName, specPath) .addParameter("deltaLakeTable", deltaTableGcsPath) .addParameter( - "deltaLakeHadoopConfig", - new org.json.JSONObject(getGcsHadoopConfig()).toString()) + "deltaLakeHadoopConfig", new org.json.JSONObject(getGcsHadoopConfig()).toString()) .addParameter("table", icebergTableIdentifier) .addParameter("catalogName", CATALOG_NAME) .addParameter( From d642c4bc93216c0c9830d73d6965332865a6b41c Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Wed, 5 Aug 2026 15:42:56 +0000 Subject: [PATCH 3/3] switch to interrupt call --- .../org/apache/beam/it/gcp/dataflow/DirectRunnerClient.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DirectRunnerClient.java b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DirectRunnerClient.java index efce55e16a..8041dc7a80 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DirectRunnerClient.java +++ b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DirectRunnerClient.java @@ -265,7 +265,7 @@ public void cancel() { currentJob.setCurrentState(JobState.CANCELLED.toString()); try { - this.stop(); + this.interrupt(); } catch (Exception e) { LOG.warn("Error cancelling job", e); }