From 1e43b219b71c5528167798e1a35e3df926f0dd17 Mon Sep 17 00:00:00 2001 From: Vitaly Terentyev Date: Thu, 22 Oct 2026 12:33:54 +0400 Subject: [PATCH 1/2] Test legacy worker --- .../beam/sdk/io/iceberg/catalog/IcebergCdcWriteBaseIT.java | 6 +++--- .../sdk/io/iceberg/cdc/sink/DeltaLakeToIcebergCdcIT.java | 6 +++--- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCdcWriteBaseIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCdcWriteBaseIT.java index d641a0e5f39e..97add476eb04 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCdcWriteBaseIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCdcWriteBaseIT.java @@ -855,9 +855,9 @@ private void roundTripThroughSource(boolean upsert, boolean changeTypeColumn) th } metadataColumns.add(IcebergCdcMetadataColumns.COMMIT_SNAPSHOT_SEQUENCE_NUMBER); PipelineOptions options = TestPipeline.testingPipelineOptions(); - if (!changeTypeColumn) { - CdcSinkTestUtils.useLegacyDataflowWorker(options); - } +// if (!changeTypeColumn) { + CdcSinkTestUtils.useLegacyDataflowWorker(options); +// } Map readConfig = new HashMap<>(managedIcebergConfig(sourceId.toString())); readConfig.put("include_metadata_columns", metadataColumns); Map writeConfig = new HashMap<>(managedIcebergConfig(targetId.toString())); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/DeltaLakeToIcebergCdcIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/DeltaLakeToIcebergCdcIT.java index b48b910b6821..bb4dcfea7e6a 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/DeltaLakeToIcebergCdcIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/DeltaLakeToIcebergCdcIT.java @@ -311,9 +311,9 @@ private void applyChangeFeed(boolean upsert, boolean changeTypeColumn) throws Ex // The native ValueKind does not survive Dataflow Runner v2 yet, so that variant runs on the // legacy worker there; the change-type column travels on any runner. - if (!changeTypeColumn) { - CdcSinkTestUtils.useLegacyDataflowWorker(p.getOptions()); - } +// if (!changeTypeColumn) { + CdcSinkTestUtils.useLegacyDataflowWorker(p.getOptions()); +// } Map readConfig = new HashMap<>(); readConfig.put("table", deltaTable); readConfig.put("start_version", 0L); From 0d8b42cb74b30c7d043ac50c267811ce567ca76f Mon Sep 17 00:00:00 2001 From: Vitaly Terentyev Date: Thu, 22 Oct 2026 13:59:46 +0400 Subject: [PATCH 2/2] Add flag for enableManagedTransforms --- sdks/java/io/iceberg/build.gradle | 1 + .../beam/sdk/io/iceberg/catalog/IcebergCdcWriteBaseIT.java | 7 ++++--- .../sdk/io/iceberg/cdc/sink/DeltaLakeToIcebergCdcIT.java | 6 +++--- 3 files changed, 8 insertions(+), 6 deletions(-) diff --git a/sdks/java/io/iceberg/build.gradle b/sdks/java/io/iceberg/build.gradle index 9c9e58edaac0..3d3b255563cc 100644 --- a/sdks/java/io/iceberg/build.gradle +++ b/sdks/java/io/iceberg/build.gradle @@ -235,6 +235,7 @@ task dataflowIntegrationTest(type: Test) { ] if (project.hasProperty('enableManagedTransforms')) { args.add("--experiments=enable_managed_transforms") + systemProperty 'beam.test.enableManagedTransforms', 'true' } systemProperty "beamTestPipelineOptions", JsonOutput.toJson(args) diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCdcWriteBaseIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCdcWriteBaseIT.java index 97add476eb04..383299f16d27 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCdcWriteBaseIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCdcWriteBaseIT.java @@ -855,9 +855,10 @@ private void roundTripThroughSource(boolean upsert, boolean changeTypeColumn) th } metadataColumns.add(IcebergCdcMetadataColumns.COMMIT_SNAPSHOT_SEQUENCE_NUMBER); PipelineOptions options = TestPipeline.testingPipelineOptions(); -// if (!changeTypeColumn) { - CdcSinkTestUtils.useLegacyDataflowWorker(options); -// } + // Use the legacy worker for managed transforms until the Runner v2 CDC issue is resolved. + if (!changeTypeColumn || Boolean.getBoolean("beam.test.enableManagedTransforms")) { + CdcSinkTestUtils.useLegacyDataflowWorker(options); + } Map readConfig = new HashMap<>(managedIcebergConfig(sourceId.toString())); readConfig.put("include_metadata_columns", metadataColumns); Map writeConfig = new HashMap<>(managedIcebergConfig(targetId.toString())); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/DeltaLakeToIcebergCdcIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/DeltaLakeToIcebergCdcIT.java index bb4dcfea7e6a..4b2a994b1e8d 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/DeltaLakeToIcebergCdcIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/DeltaLakeToIcebergCdcIT.java @@ -311,9 +311,9 @@ private void applyChangeFeed(boolean upsert, boolean changeTypeColumn) throws Ex // The native ValueKind does not survive Dataflow Runner v2 yet, so that variant runs on the // legacy worker there; the change-type column travels on any runner. -// if (!changeTypeColumn) { - CdcSinkTestUtils.useLegacyDataflowWorker(p.getOptions()); -// } + if (!changeTypeColumn || Boolean.getBoolean("beam.test.enableManagedTransforms")) { + CdcSinkTestUtils.useLegacyDataflowWorker(p.getOptions()); + } Map readConfig = new HashMap<>(); readConfig.put("table", deltaTable); readConfig.put("start_version", 0L);