[DebeziumIO] Upgrade to Debezium 3.5.2.Final

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/DebeziumIO.java

@@ -36,7 +36,6 @@
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Joiner;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
-import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Maps;
 import org.apache.kafka.connect.source.SourceConnector;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.checkerframework.checker.nullness.qual.Nullable;
@@ -75,7 +74,7 @@
  *             .withConnectorClass(MySqlConnector.class)
  *             .withConnectionProperty("database.server.id", "184054")
  *             .withConnectionProperty("database.server.name", "serverid")
- *             .withConnectionProperty("database.history", DebeziumSDFDatabaseHistory.class.getName())
+ *             .withConnectionProperty("schema.history.internal", DebeziumSDFDatabaseHistory.class.getName())
  *             .withConnectionProperty("include.schema.changes", "false");
  *
  *      PipelineOptions options = PipelineOptionsFactory.create();
@@ -313,9 +312,9 @@
           new KafkaSourceConsumerFn.OffsetTracker(
               new KafkaSourceConsumerFn.OffsetHolder(null, null, 0)));
 
-      Map<String, String> connectorConfig =
-          Maps.newHashMap(getConnectorConfiguration().getConfigurationMap());
-      connectorConfig.put("snapshot.mode", "schema_only");
+      // Deliberately runs with the connector's configured snapshot mode: schema inference samples
+      // an actual data record, which a schema-only snapshot ("no_data", formerly "schema_only")
+      // would never emit.
       SourceRecord sampledRecord =
           fn.getOneRecord(getConnectorConfiguration().getConfigurationMap());
       fn.reset();
@@ -401,9 +400,9 @@
      */
     public static ConnectorConfiguration create() {
       return new AutoValue_DebeziumIO_ConnectorConfiguration.Builder()
           .setConnectionProperties(ValueProvider.StaticValueProvider.of(new HashMap<>()))
           .build();
     }
 
     /**
      * Applies the connectorClass to be used to connect to your database.
@@ -420,9 +419,9 @@
      * @return {@link ConnectorConfiguration}
      */
     public ConnectorConfiguration withConnectorClass(Class<?> connectorClass) {
       checkArgument(connectorClass != null, "connectorClass can not be null");
       return withConnectorClass(ValueProvider.StaticValueProvider.of(connectorClass));
     }
 
     /**
      * Sets the connectorClass to be used to connect to your database. It cannot be null.
@@ -641,10 +640,10 @@
         configuration.computeIfAbsent(entry.getKey(), k -> entry.getValue());
       }
 
-      // Set default Database History impl. if not provided implementation and Kafka topic prefix,
-      // if not provided
+      // Set default schema history impl. if not provided implementation and Kafka topic prefix,
+      // if not provided. Before Debezium 2.0 this key was named "database.history".
       configuration.computeIfAbsent(
-          "database.history",
+          "schema.history.internal",
           k -> KafkaSourceConsumerFn.DebeziumSDFDatabaseHistory.class.getName());
       configuration.computeIfAbsent("topic.prefix", k -> "beam-debezium-connector");
       configuration.computeIfAbsent(

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/DebeziumIOTest.java

@@ -52,7 +52,8 @@ public class DebeziumIOTest implements Serializable {
           .withConnectionProperty("database.server.id", "184054")
           .withConnectionProperty("database.server.name", "dbserver1")
           .withConnectionProperty(
-              "database.history", KafkaSourceConsumerFn.DebeziumSDFDatabaseHistory.class.getName())
+              "schema.history.internal",
+              KafkaSourceConsumerFn.DebeziumSDFDatabaseHistory.class.getName())
           .withConnectionProperty("include.schema.changes", "false");
 
   @Test

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",