[IO] Fix Python Transform and IT for DebeziumIO

Author: tkaymak

sdks/java/io/debezium/build.gradle

@@ -36,6 +36,11 @@ dependencies {
     implementation project(path: ":sdks:java:core", configuration: "shadow")
     implementation library.java.slf4j_api
     implementation library.java.joda_time
+    // Explicitly declare Jackson dependencies
+    // and align with the 2.17.1 version.
+    implementation 'com.fasterxml.jackson.core:jackson-core:2.17.1'
+    implementation 'com.fasterxml.jackson.core:jackson-databind:2.17.1'
+
     provided library.java.jackson_dataformat_csv
     permitUnusedDeclared library.java.jackson_dataformat_csv
 
@@ -66,26 +71,21 @@ dependencies {
     testImplementation group: 'io.debezium', name: 'debezium-connector-postgres', version: '3.1.1.Final'
 }
 
-// TODO: Remove pin after Beam has unpinned it (PR #35231)
-// Pin the Antlr version
+// Pin the Antlr version to 4.10
+// and force Jackson versions to 2.17.1
+// TODO: Remove pin after upgrading Beam's Jackson version
+// This overrides the global 2.15.4 forced by BeamModulePlugin.
 configurations.all {
     resolutionStrategy {
-        force 'org.antlr:antlr4:4.10', 'org.antlr:antlr4-runtime:4.10'
+        force 'org.antlr:antlr4:4.10',
+              'org.antlr:antlr4-runtime:4.10',
+              'com.fasterxml.jackson.core:jackson-core:2.17.1',
+              'com.fasterxml.jackson.core:jackson-annotations:2.17.1',
+              'com.fasterxml.jackson.core:jackson-databind:2.17.1',
+              'com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.17.1'
     }
 }
 
-// TODO: Remove pin after upgrading Beam's Jackson version
-// Force Jackson versions for the test runtime classpath
-configurations.named("testRuntimeClasspath") {
-    resolutionStrategy.force (
-        'com.fasterxml.jackson.core:jackson-core:2.17.1',
-        'com.fasterxml.jackson.core:jackson-annotations:2.17.1',
-        'com.fasterxml.jackson.core:jackson-databind:2.17.1',
-        'com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.17.1',
-        'com.fasterxml.jackson.module:jackson-module-afterburner:2.17.1'
-    )
-}
-
 def configureTestJvmArgs(Task task) {
     List<String> currentJvmArgs = task.jvmArgs ? new ArrayList<>(task.jvmArgs) : new ArrayList<>()
 

sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java

@@ -17,6 +17,8 @@
  */
 package org.apache.beam.io.debezium;
 
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.auto.service.AutoService;
 import com.google.auto.value.AutoValue;
 import java.util.Arrays;
@@ -121,8 +123,23 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) {
                   configuration.getTable().substring(0, configuration.getTable().indexOf(".")));
         }
 
-        final List<String> debeziumConnectionProperties =
-            configuration.getDebeziumConnectionProperties();
+        // Logic to handle both JSON string and list of strings from Portability Framework Runners
+        List<String> debeziumConnectionProperties = configuration.getDebeziumConnectionProperties();
+        final String jsonProperties = configuration.getDebeziumConnectionPropertiesJson();
+
+        // Check for the new JSON properties field first.
+        if (jsonProperties != null) {
+          try {
+            ObjectMapper mapper = new ObjectMapper();
+            debeziumConnectionProperties =
+                mapper.readValue(jsonProperties, new TypeReference<List<String>>() {});
+          } catch (Exception e) {
+            throw new IllegalArgumentException(
+                "Unable to parse debeziumConnectionPropertiesJson", e);
+          }
+        }
+
+        // Fall back to Java list-based properties if JSON is empty
         if (debeziumConnectionProperties != null) {
           for (String connectionProperty : debeziumConnectionProperties) {
             String[] parts = connectionProperty.split("=", -1);
@@ -190,6 +207,8 @@ public abstract static class DebeziumReadSchemaTransformConfiguration {
 
     public abstract @Nullable List<String> getDebeziumConnectionProperties();
 
+    public abstract @Nullable String getDebeziumConnectionPropertiesJson();
+
     public static Builder builder() {
       return new AutoValue_DebeziumReadSchemaTransformProvider_DebeziumReadSchemaTransformConfiguration
           .Builder();
@@ -212,6 +231,9 @@ public abstract static class Builder {
       public abstract Builder setDebeziumConnectionProperties(
           List<String> debeziumConnectionProperties);
 
+      public abstract Builder setDebeziumConnectionPropertiesJson(
+          String debeziumConnectionPropertiesJson);
+
       public abstract DebeziumReadSchemaTransformConfiguration build();
     }
   }

sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumTransformRegistrar.java

@@ -17,7 +17,10 @@
  */
 package org.apache.beam.io.debezium;
 
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.auto.service.AutoService;
+import java.io.IOException;
 import java.util.List;
 import java.util.Map;
 import org.apache.beam.sdk.expansion.ExternalTransformRegistrar;
@@ -76,12 +79,17 @@ public static class ReadBuilder
 
     public static class Configuration extends CrossLanguageConfiguration {
       private @Nullable List<String> connectionProperties;
+      private @Nullable String connectionPropertiesJson;
       private @Nullable Long maxNumberOfRecords;
 
       public void setConnectionProperties(@Nullable List<String> connectionProperties) {
         this.connectionProperties = connectionProperties;
       }
 
+      public void setConnectionPropertiesJson(@Nullable String connectionPropertiesJson) {
+        this.connectionPropertiesJson = connectionPropertiesJson;
+      }
+
       public void setMaxNumberOfRecords(@Nullable Long maxNumberOfRecords) {
         this.maxNumberOfRecords = maxNumberOfRecords;
       }
@@ -97,12 +105,29 @@ public PTransform<PBegin, PCollection<String>> buildExternal(Configuration confi
               .withPort(configuration.port)
               .withConnectorClass(configuration.connectorClass.getConnector());
 
-      if (configuration.connectionProperties != null) {
-        for (String connectionProperty : configuration.connectionProperties) {
+      List<String> propertiesToProcess = null;
+      if (configuration.connectionPropertiesJson != null) {
+        // Prioritize the new JSON format if present
+        try {
+          ObjectMapper mapper = new ObjectMapper();
+          propertiesToProcess =
+              mapper.readValue(
+                  configuration.connectionPropertiesJson, new TypeReference<List<String>>() {});
+        } catch (IOException e) {
+          throw new IllegalArgumentException("Error parsing connectionPropertiesJson.", e);
+        }
+      } else {
+        // Fall back to the old list format for backward-compatibility
+        propertiesToProcess = configuration.connectionProperties;
+      }
+
+      if (propertiesToProcess != null) {
+        for (String connectionProperty : propertiesToProcess) {
           String[] parts = connectionProperty.split("=", -1);
-          String key = parts[0];
-          String value = parts[1];
-          connectorConfiguration.withConnectionProperty(key, value);
+          if (parts.length == 2) {
+            connectorConfiguration =
+                connectorConfiguration.withConnectionProperty(parts[0], parts[1]);
+          }
         }
       }
 

sdks/python/apache_beam/io/debezium.py

@@ -80,12 +80,10 @@
 
 import json
 from enum import Enum
-from typing import List
 from typing import NamedTuple
 from typing import Optional
 
-from apache_beam.transforms import DoFn
-from apache_beam.transforms import ParDo
+from apache_beam.transforms import Map
 from apache_beam.transforms import PTransform
 from apache_beam.transforms.external import BeamJarExpansionService
 from apache_beam.transforms.external import ExternalTransform
@@ -110,14 +108,7 @@ class DriverClassName(Enum):
     'ReadFromDebeziumSchema',
     [('connector_class', str), ('username', str), ('password', str),
      ('host', str), ('port', str), ('max_number_of_records', Optional[int]),
-     ('connection_properties', List[str])])
-
-
-class _JsonStringToDictionaries(DoFn):
-  """ A DoFn that consumes a JSON string and yields a python dictionary """
-  def process(self, json_string):
-    obj = json.loads(json_string)
-    yield obj
+     ('connection_properties_json', str)])
 
 
 class ReadFromDebezium(PTransform):
@@ -152,19 +143,22 @@ def __init__(
                                       to be fetched before stop.
         :param connection_properties: properties of the debezium
                                       connection passed as string
-                                      with format
-                                      [propertyName=property;]*
+                                      with json format
+                                      {"propertyName": "property"}
         :param expansion_service: The address (host:port)
                                   of the ExpansionService.
     """
+    serialized_properties = json.dumps(
+        connection_properties) if connection_properties else "[]"
+
     self.params = ReadFromDebeziumSchema(
         connector_class=connector_class.value,
         username=username,
         password=password,
         host=host,
         port=port,
         max_number_of_records=max_number_of_records,
-        connection_properties=connection_properties)
+        connection_properties_json=serialized_properties)
     self.expansion_service = expansion_service or default_io_expansion_service()
 
   def expand(self, pbegin):
@@ -173,4 +167,5 @@ def expand(self, pbegin):
             self.URN,
             NamedTupleBasedPayloadBuilder(self.params),
             self.expansion_service,
-        ) | ParDo(_JsonStringToDictionaries()))
+        ).with_output_types(str)
+        | 'JsonToDict' >> Map(json.loads).with_output_types(dict))

sdks/python/apache_beam/io/external/xlang_debeziumio_it_test.py

@@ -74,7 +74,6 @@ def setUp(self):
     self.connection_properties = [
         "database.dbname=inventory",
         "database.server.name=dbserver1",
-        "database.include.list=inventory",
         "include.schema.changes=false"
     ]
 
@@ -90,8 +89,8 @@ def test_xlang_debezium_read(self):
     expected_response = [{
         "metadata": {
             "connector": "postgresql",
-            "version": "1.3.1.Final",
-            "name": "dbserver1",
+            "version": "3.1.1.Final",
+            "name": "beam-debezium-connector",
             "database": "inventory",
             "schema": "inventory",
             "table": "customers"