(beam) branch fix-managed-iceberg updated: Add flag for enableManagedTransforms

[vterentev]

This is an automated email from the ASF dual-hosted git repository.

Amar3tto pushed a commit to branch fix-managed-iceberg
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/fix-managed-iceberg by this 
push:
     new 0d8b42cb74b Add flag for enableManagedTransforms
0d8b42cb74b is described below

commit 0d8b42cb74b30c7d043ac50c267811ce567ca76f
Author: Vitaly Terentyev <vitaly.terent...@akvelon.com>
AuthorDate: Thu Oct 22 13:59:46 2026 +0400

    Add flag for enableManagedTransforms
---
 sdks/java/io/iceberg/build.gradle                                  | 1 +
 .../apache/beam/sdk/io/iceberg/catalog/IcebergCdcWriteBaseIT.java  | 7 ++++---
 .../beam/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 9c9e58edaac..3d3b255563c 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 97add476eb0..383299f16d2 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 @@ public abstract class IcebergCdcWriteBaseIT implements 
Serializable {
     }
     
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<String, Object> readConfig = new 
HashMap<>(managedIcebergConfig(sourceId.toString()));
     readConfig.put("include_metadata_columns", metadataColumns);
     Map<String, Object> 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 bb4dcfea7e6..4b2a994b1e8 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 @@ public class DeltaLakeToIcebergCdcIT {
 
     // 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<String, Object> readConfig = new HashMap<>();
     readConfig.put("table", deltaTable);
     readConfig.put("start_version", 0L);