try only pin dependency

Author: Abacn

sdks/java/io/debezium/build.gradle

@@ -40,7 +40,8 @@ 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'
-
+    permitUnusedDeclared 'com.fasterxml.jackson.core:jackson-core:2.17.1'
+    permitUnusedDeclared 'com.fasterxml.jackson.core:jackson-databind:2.17.1'
     provided library.java.jackson_dataformat_csv
     permitUnusedDeclared library.java.jackson_dataformat_csv
 
@@ -78,11 +79,11 @@ dependencies {
 configurations.all {
     resolutionStrategy {
         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'
+                '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'
     }
 }
 

sdks/java/io/debezium/expansion-service/build.gradle

@@ -46,3 +46,21 @@ dependencies {
     runtimeOnly group: 'io.debezium', name: 'debezium-connector-oracle', version: debezium_version
     runtimeOnly group: 'io.debezium', name: 'debezium-connector-db2', version: debezium_version
 }
+
+// sync with debezium/build.gradle depenency pins
+configurations.all {
+    resolutionStrategy {
+        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'
+    }
+}
+
+shadowJar {
+    manifest {
+        attributes(["Multi-Release": true])
+    }
+}

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

@@ -17,8 +17,6 @@
  */
 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;
@@ -123,23 +121,8 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) {
                   configuration.getTable().substring(0, configuration.getTable().indexOf(".")));
         }
 
-        // 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
+        final List<String> debeziumConnectionProperties =
+            configuration.getDebeziumConnectionProperties();
         if (debeziumConnectionProperties != null) {
           for (String connectionProperty : debeziumConnectionProperties) {
             String[] parts = connectionProperty.split("=", -1);
@@ -207,8 +190,6 @@ 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();
@@ -231,9 +212,6 @@ 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,10 +17,7 @@
  */
 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;
@@ -79,17 +76,12 @@ 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;
       }
@@ -105,29 +97,12 @@ public PTransform<PBegin, PCollection<String>> buildExternal(Configuration confi
               .withPort(configuration.port)
               .withConnectorClass(configuration.connectorClass.getConnector());
 
-      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) {
+      if (configuration.connectionProperties != null) {
+        for (String connectionProperty : configuration.connectionProperties) {
           String[] parts = connectionProperty.split("=", -1);
-          if (parts.length == 2) {
-            connectorConfiguration =
-                connectorConfiguration.withConnectionProperty(parts[0], parts[1]);
-          }
+          String key = parts[0];
+          String value = parts[1];
+          connectorConfiguration = connectorConfiguration.withConnectionProperty(key, value);
         }
       }
 

sdks/python/apache_beam/io/debezium.py

@@ -80,6 +80,7 @@
 
 import json
 from enum import Enum
+from typing import List
 from typing import NamedTuple
 from typing import Optional
 
@@ -108,7 +109,7 @@ class DriverClassName(Enum):
     'ReadFromDebeziumSchema',
     [('connector_class', str), ('username', str), ('password', str),
      ('host', str), ('port', str), ('max_number_of_records', Optional[int]),
-     ('connection_properties_json', str)])
+     ('connection_properties', List[str])])
 
 
 class ReadFromDebezium(PTransform):
@@ -143,13 +144,11 @@ def __init__(
                                       to be fetched before stop.
         :param connection_properties: properties of the debezium
                                       connection passed as string
-                                      with json format
-                                      {"propertyName": "property"}
+                                      with with 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,
@@ -158,7 +157,7 @@ def __init__(
         host=host,
         port=port,
         max_number_of_records=max_number_of_records,
-        connection_properties_json=serialized_properties)
+        connection_properties=connection_properties)
     self.expansion_service = expansion_service or default_io_expansion_service()
 
   def expand(self, pbegin):

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

@@ -129,7 +129,7 @@ def start_db_container(self, retries):
       try:
         self.db = PostgresContainer(
             'quay.io/debezium/example-postgres:latest',
-            user=self.username,
+            username=self.username,
             password=self.password,
             dbname=self.database)
         self.db.start()