Upgrade DebeziumIO to Debezium 3.5.2.Final

Debezium 3.5 builds against Kafka Connect 4.1.2, so the connect-api pin
moves from 3.9.0 to 4.1.2 and kafka-clients is declared and forced to the
same version. Connect 4.1 added an abstract pluginMetrics() to
SourceTaskContext; BeamSourceTaskContext returns null, which is safe here
because no Debezium 3.5.2 connector calls it.

Debezium 3.5 split debezium-core into smaller modules, so debezium-config
and debezium-connector-common are now declared explicitly. The
expansion-service shadow jar was checked to still bundle the split
classes and all five connectors.

The Postgres connector needs JDBC driver 42.7.x, so 42.7.7 is forced in
this module only, matching the upstream Debezium 3.5.2 pin. The Beam-wide
42.6.2 pin from 24f562d is deliberately left untouched.

Test updates for 3.5 behaviour changes: connection failures now surface
wrapped in a RetriableException, so the schema transform tests unwrap the
cause chain, and the example-mysql image now names its binlog
mysql-bin.000003 instead of binlog.000002.

3.6.0 was skipped for now because of open MySQL DDL parser regressions in
that line.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: tkaymak

sdks/go/test/integration/io/xlang/debezium/debezium.go

@@ -31,7 +31,7 @@ func ReadPipeline(addr, username, password, dbname, host, port string, connector
 		connectorClass, reflectx.String, debeziumio.MaxRecord(maxrecords),
 		debeziumio.MaxTimeToRun(120000),
 		debeziumio.ConnectionProperties(connectionProperties), debeziumio.ExpansionAddr(addr))
-	expectedJson := `{"metadata":{"connector":"postgresql","version":"3.1.3.Final","name":"beam-debezium-connector","database":"inventory","schema":"inventory","table":"customers"},"before":null,"after":{"fields":{"last_name":"Thomas","id":1001,"first_name":"Sally","email":"sally.thomas@acme.com"}}}`
+	expectedJson := `{"metadata":{"connector":"postgresql","version":"3.5.2.Final","name":"beam-debezium-connector","database":"inventory","schema":"inventory","table":"customers"},"before":null,"after":{"fields":{"last_name":"Thomas","id":1001,"first_name":"Sally","email":"sally.thomas@acme.com"}}}`
 	expected := beam.Create(s, expectedJson)
 	passert.Equals(s, result, expected)
 	return p

sdks/go/test/integration/io/xlang/debezium/debezium_test.go

@@ -33,7 +33,7 @@ import (
 )
 
 const (
-	debeziumImage = "quay.io/debezium/example-postgres:3.1.3.Final"
+	debeziumImage = "quay.io/debezium/example-postgres:3.5.2.Final"
 	debeziumPort  = "5432/tcp"
 	maxRetries    = 5
 )

sdks/java/io/debezium/build.gradle

@@ -46,10 +46,13 @@ dependencies {
     permitUnusedDeclared library.java.jackson_dataformat_csv
 
     // Kafka connect dependencies
-    implementation "org.apache.kafka:connect-api:3.9.0"
+    implementation "org.apache.kafka:connect-api:4.1.2"
+    implementation "org.apache.kafka:kafka-clients:4.1.2"
 
     // Debezium dependencies
-    implementation group: 'io.debezium', name: 'debezium-core', version: '3.1.3.Final'
+    implementation group: 'io.debezium', name: 'debezium-core', version: '3.5.2.Final'
+    implementation group: 'io.debezium', name: 'debezium-config', version: '3.5.2.Final'
+    implementation group: 'io.debezium', name: 'debezium-connector-common', version: '3.5.2.Final'
 
     // Test dependencies
     testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
@@ -64,12 +67,12 @@ dependencies {
     testImplementation "org.testcontainers:kafka"
     testImplementation "org.testcontainers:mysql"
     testImplementation "org.testcontainers:postgresql"
-    testImplementation "io.debezium:debezium-testing-testcontainers:3.1.3.Final"
+    testImplementation "io.debezium:debezium-testing-testcontainers:3.5.2.Final"
     testImplementation 'com.zaxxer:HikariCP:5.1.0'
 
     // Debezium connector implementations for testing
-    testImplementation group: 'io.debezium', name: 'debezium-connector-mysql', version: '3.1.3.Final'
-    testImplementation group: 'io.debezium', name: 'debezium-connector-postgres', version: '3.1.3.Final'
+    testImplementation group: 'io.debezium', name: 'debezium-connector-mysql', version: '3.5.2.Final'
+    testImplementation group: 'io.debezium', name: 'debezium-connector-postgres', version: '3.5.2.Final'
 }
 
 // Pin the Antlr version to 4.10
@@ -83,7 +86,9 @@ configurations.all {
                 '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.datatype:jackson-datatype-jsr310:2.17.1',
+                'org.apache.kafka:kafka-clients:4.1.2',
+                'org.postgresql:postgresql:42.7.7'
     }
 }
 

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

@@ -39,7 +39,7 @@ dependencies {
     runtimeOnly library.java.slf4j_jdk14
 
     // Debezium runtime dependencies
-    def debezium_version = '3.1.3.Final'
+    def debezium_version = '3.5.2.Final'
     runtimeOnly group: 'io.debezium', name: 'debezium-connector-mysql', version: debezium_version
     runtimeOnly group: 'io.debezium', name: 'debezium-connector-postgres', version: debezium_version
     runtimeOnly group: 'io.debezium', name: 'debezium-connector-sqlserver', version: debezium_version
@@ -55,7 +55,9 @@ configurations.all {
                 '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.datatype:jackson-datatype-jsr310:2.17.1',
+                'org.apache.kafka:kafka-clients:4.1.2',
+                'org.postgresql:postgresql:42.7.7'
     }
 }
 

sdks/java/io/debezium/src/README.md

@@ -25,7 +25,7 @@ DebeziumIO is an Apache Beam connector that lets users connect their Events-Driv
 
 ### Getting Started
 
-DebeziumIO uses [Debezium Connectors v3.1](https://debezium.io/documentation/reference/3.1/connectors/) to connect to Apache Beam. All you need to do is choose the Debezium Connector that suits your Debezium setup and pick a [Serializable Function](https://beam.apache.org/releases/javadoc/2.65.0/org/apache/beam/sdk/transforms/SerializableFunction.html), then you will be able to connect to Apache Beam and start building your own Pipelines.
+DebeziumIO uses [Debezium Connectors v3.5](https://debezium.io/documentation/reference/3.5/connectors/) to connect to Apache Beam. All you need to do is choose the Debezium Connector that suits your Debezium setup and pick a [Serializable Function](https://beam.apache.org/releases/javadoc/2.65.0/org/apache/beam/sdk/transforms/SerializableFunction.html), then you will be able to connect to Apache Beam and start building your own Pipelines.
 
 These connectors have been successfully tested and are known to work fine:
 *  MySQL Connector
@@ -65,7 +65,7 @@ You can also add more configuration, such as Connector-specific Properties with
 |Method|Params|Description|
 |-|-|-|
 |`.withConnectionProperty(propName, propValue)`|_String_, _String_|Adds a custom property to the connector.|
-> **Note:** For more information on custom properties, see your [Debezium Connector](https://debezium.io/documentation/reference/3.1/connectors/) specific documentation.
+> **Note:** For more information on custom properties, see your [Debezium Connector](https://debezium.io/documentation/reference/3.5/connectors/) specific documentation.
 
 Example of a MySQL Debezium Connector setup:
 ```
@@ -160,7 +160,7 @@ By default, DebeziumIO initializes it with the former, though user may choose th
 ### Requirements and Supported versions
 
 -  JDK v17
--  Debezium Connectors v3.1
+-  Debezium Connectors v3.5
 -  Apache Beam 2.66
 
 ## Running Unit Tests

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

@@ -350,6 +350,11 @@ public OffsetStorageReader offsetStorageReader() {
       LOG.debug("------------- Creating an offset storage reader");
       return new DebeziumSourceOffsetStorageReader(initialOffset);
     }
+
+    @Override
+    public org.apache.kafka.common.metrics.PluginMetrics pluginMetrics() {
+      return null;
+    }
   }
 
   private static class DebeziumSourceOffsetStorageReader implements OffsetStorageReader {

sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOMySqlConnectorIT.java

@@ -74,7 +74,7 @@ public class DebeziumIOMySqlConnectorIT {
   @ClassRule
   public static final MySQLContainer<?> MY_SQL_CONTAINER =
       new MySQLContainer<>(
-              DockerImageName.parse("quay.io/debezium/example-mysql:3.1.3.Final")
+              DockerImageName.parse("quay.io/debezium/example-mysql:3.5.2.Final")
                   .asCompatibleSubstituteFor("mysql"))
           .withPassword("debezium")
           .withUsername("mysqluser")
@@ -277,8 +277,8 @@ public void testDebeziumIOMySql() {
                 .withMaxNumberOfRecords(30)
                 .withCoder(StringUtf8Coder.of()));
     String expected =
-        "{\"metadata\":{\"connector\":\"mysql\",\"version\":\"3.1.3.Final\",\"name\":\"beam-debezium-connector\","
-            + "\"database\":\"inventory\",\"schema\":\"binlog.000002\",\"table\":\"addresses\"},\"before\":null,"
+        "{\"metadata\":{\"connector\":\"mysql\",\"version\":\"3.5.2.Final\",\"name\":\"beam-debezium-connector\","
+            + "\"database\":\"inventory\",\"schema\":\"mysql-bin.000003\",\"table\":\"addresses\"},\"before\":null,"
             + "\"after\":{\"fields\":{\"zip\":\"76036\",\"city\":\"Euless\","
             + "\"street\":\"3183 Moore Avenue\",\"id\":10,\"state\":\"Texas\",\"customer_id\":1001,"
             + "\"type\":\"SHIPPING\"}}}";

sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java

@@ -56,7 +56,7 @@ public class DebeziumIOPostgresSqlConnectorIT {
   @ClassRule
   public static final PostgreSQLContainer<?> POSTGRES_SQL_CONTAINER =
       new PostgreSQLContainer<>(
-              DockerImageName.parse("quay.io/debezium/example-postgres:3.1.3.Final")
+              DockerImageName.parse("quay.io/debezium/example-postgres:3.5.2.Final")
                   .asCompatibleSubstituteFor("postgres"))
           .withPassword("dbz")
           .withUsername("debezium")
@@ -180,7 +180,7 @@ public void testDebeziumIOPostgresSql() {
                 .withMaxNumberOfRecords(30)
                 .withCoder(StringUtf8Coder.of()));
     String expected =
-        "{\"metadata\":{\"connector\":\"postgresql\",\"version\":\"3.1.3.Final\",\"name\":\"beam-debezium-connector\","
+        "{\"metadata\":{\"connector\":\"postgresql\",\"version\":\"3.5.2.Final\",\"name\":\"beam-debezium-connector\","
             + "\"database\":\"inventory\",\"schema\":\"inventory\",\"table\":\"customers\"},\"before\":null,"
             + "\"after\":{\"fields\":{\"last_name\":\"Thomas\",\"id\":1001,\"first_name\":\"Sally\","
             + "\"email\":\"sally.thomas@acme.com\"}}}";

sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformTest.java

@@ -18,6 +18,7 @@
 package org.apache.beam.io.debezium;
 
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThrows;
 
 import io.debezium.DebeziumException;
@@ -60,7 +61,7 @@ public class DebeziumReadSchemaTransformTest {
   @ClassRule
   public static final MySQLContainer<?> MY_SQL_CONTAINER =
       new MySQLContainer<>(
-              DockerImageName.parse("debezium/example-mysql:1.4")
+              DockerImageName.parse("quay.io/debezium/example-mysql:3.5.2.Final")
                   .asCompatibleSubstituteFor("mysql"))
           .withPassword("debezium")
           .withUsername("mysqluser")
@@ -118,6 +119,17 @@ private PTransform<PCollectionRowTuple, PCollectionRowTuple> makePtransform(
                 .build());
   }
 
+  // Since Debezium 3.5 connection failures surface as a RetriableException wrapping the
+  // DebeziumException instead of a top-level DebeziumException.
+  private static DebeziumException findDebeziumException(Throwable thrown) {
+    Throwable cause = thrown;
+    while (cause != null && !(cause instanceof DebeziumException)) {
+      cause = cause.getCause();
+    }
+    assertNotNull("Expected DebeziumException in cause chain", cause);
+    return (DebeziumException) cause;
+  }
+
   @Test
   public void testNoProblem() {
     Pipeline readPipeline = Pipeline.create();
@@ -142,9 +154,9 @@ public void testNoProblem() {
   @Test
   public void testWrongUser() {
     Pipeline readPipeline = Pipeline.create();
-    DebeziumException ex =
+    Exception thrown =
         assertThrows(
-            DebeziumException.class,
+            Exception.class,
             () -> {
               PCollectionRowTuple.empty(readPipeline)
                   .apply(
@@ -156,16 +168,17 @@ public void testWrongUser() {
                           "localhost"))
                   .get("output");
             });
+    DebeziumException ex = findDebeziumException(thrown);
     assertThat(ex.getCause().getMessage(), Matchers.containsString("password"));
     assertThat(ex.getCause().getMessage(), Matchers.containsString("wrongUser"));
   }
 
   @Test
   public void testWrongPassword() {
     Pipeline readPipeline = Pipeline.create();
-    DebeziumException ex =
+    Exception thrown =
         assertThrows(
-            DebeziumException.class,
+            Exception.class,
             () -> {
               PCollectionRowTuple.empty(readPipeline)
                   .apply(
@@ -177,21 +190,23 @@ public void testWrongPassword() {
                           "localhost"))
                   .get("output");
             });
+    DebeziumException ex = findDebeziumException(thrown);
     assertThat(ex.getCause().getMessage(), Matchers.containsString("password"));
     assertThat(ex.getCause().getMessage(), Matchers.containsString(userName));
   }
 
   @Test
   public void testWrongPort() {
     Pipeline readPipeline = Pipeline.create();
-    DebeziumException ex =
+    Exception thrown =
         assertThrows(
-            DebeziumException.class,
+            Exception.class,
             () -> {
               PCollectionRowTuple.empty(readPipeline)
                   .apply(makePtransform(userName, password, database, 12345, "localhost"))
                   .get("output");
             });
+    DebeziumException ex = findDebeziumException(thrown);
     Throwable lowestCause = ex.getCause();
     while (lowestCause.getCause() != null) {
       lowestCause = lowestCause.getCause();

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

@@ -89,7 +89,7 @@ def test_xlang_debezium_read(self):
     expected_response = [{
         "metadata": {
             "connector": "postgresql",
-            "version": "3.1.3.Final",
+            "version": "3.5.2.Final",
             "name": "beam-debezium-connector",
             "database": "inventory",
             "schema": "inventory",