diff --git a/sdks/go/test/integration/io/xlang/debezium/debezium.go b/sdks/go/test/integration/io/xlang/debezium/debezium.go index e1b9bab963c3..a4046f018740 100644 --- a/sdks/go/test/integration/io/xlang/debezium/debezium.go +++ b/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 diff --git a/sdks/go/test/integration/io/xlang/debezium/debezium_test.go b/sdks/go/test/integration/io/xlang/debezium/debezium_test.go index 8ccb64cae209..b234dd5d8f8a 100644 --- a/sdks/go/test/integration/io/xlang/debezium/debezium_test.go +++ b/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 ) diff --git a/sdks/java/io/debezium/build.gradle b/sdks/java/io/debezium/build.gradle index c488ac17d990..ad5410b37dee 100644 --- a/sdks/java/io/debezium/build.gradle +++ b/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' } } diff --git a/sdks/java/io/debezium/expansion-service/build.gradle b/sdks/java/io/debezium/expansion-service/build.gradle index 82a34b5c0665..fe820b62c138 100644 --- a/sdks/java/io/debezium/expansion-service/build.gradle +++ b/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' } } diff --git a/sdks/java/io/debezium/src/README.md b/sdks/java/io/debezium/src/README.md index 535213218856..6be7294fa499 100644 --- a/sdks/java/io/debezium/src/README.md +++ b/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 diff --git a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumIO.java b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumIO.java index 6c31d5a02349..b89b3644f615 100644 --- a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumIO.java +++ b/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 @@ protected Schema getRecordSchema() { new KafkaSourceConsumerFn.OffsetTracker( new KafkaSourceConsumerFn.OffsetHolder(null, null, 0))); - Map 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(); @@ -641,10 +640,10 @@ public Map getConfigurationMap() { 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( diff --git a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java index d298ddd9cafb..89fc2a5c085d 100644 --- a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java +++ b/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 { diff --git a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOMySqlConnectorIT.java b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOMySqlConnectorIT.java index 3fe86a29cce5..e0f811a8495d 100644 --- a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOMySqlConnectorIT.java +++ b/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\"}}}"; diff --git a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java index 87b9bbb92e5e..0a76415d23ba 100644 --- a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java +++ b/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\"}}}"; diff --git a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOTest.java b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOTest.java index 80509f5bb911..074f0ba5b526 100644 --- a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOTest.java +++ b/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 diff --git a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformTest.java b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformTest.java index 2fc8996ba55e..b961ad84a7b8 100644 --- a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformTest.java +++ b/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 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,6 +168,7 @@ public void testWrongUser() { "localhost")) .get("output"); }); + DebeziumException ex = findDebeziumException(thrown); assertThat(ex.getCause().getMessage(), Matchers.containsString("password")); assertThat(ex.getCause().getMessage(), Matchers.containsString("wrongUser")); } @@ -163,9 +176,9 @@ public void testWrongUser() { @Test public void testWrongPassword() { Pipeline readPipeline = Pipeline.create(); - DebeziumException ex = + Exception thrown = assertThrows( - DebeziumException.class, + Exception.class, () -> { PCollectionRowTuple.empty(readPipeline) .apply( @@ -177,6 +190,7 @@ public void testWrongPassword() { "localhost")) .get("output"); }); + DebeziumException ex = findDebeziumException(thrown); assertThat(ex.getCause().getMessage(), Matchers.containsString("password")); assertThat(ex.getCause().getMessage(), Matchers.containsString(userName)); } @@ -184,14 +198,15 @@ public void testWrongPassword() { @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(); diff --git a/sdks/python/apache_beam/io/external/xlang_debeziumio_it_test.py b/sdks/python/apache_beam/io/external/xlang_debeziumio_it_test.py index 30b96f01a1a5..5fc7c1567cd9 100644 --- a/sdks/python/apache_beam/io/external/xlang_debeziumio_it_test.py +++ b/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",