From a9ae7c590bbc956fdae9fb2b3bc3d9e4dc64a38a Mon Sep 17 00:00:00 2001 From: Tobias Kaymak Date: Fri, 31 Jul 2026 10:27:57 +0000 Subject: [PATCH 1/3] Remove dead snapshot.mode override in DebeziumIO.getRecordSchema getRecordSchema copied the connector configuration, set snapshot.mode on the copy, then passed the original unmodified map to getOneRecord. The override therefore never took effect, and schema inference has always run with the connector's configured snapshot mode. Drop the dead copy rather than start honouring it. getOneRecord derives the schema by sampling an actual data record, so a schema-only snapshot would emit nothing and schema inference would fail with "could not fetch database schema". No behaviour change. --- .../main/java/org/apache/beam/io/debezium/DebeziumIO.java | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) 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..1f967ba8316a 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; @@ -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(); From 897f5c001140d778387a500e6ec6c90d590e30f1 Mon Sep 17 00:00:00 2001 From: Tobias Kaymak Date: Fri, 31 Jul 2026 10:28:12 +0000 Subject: [PATCH 2/3] 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 24f562d84f6 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. --- .../integration/io/xlang/debezium/debezium.go | 2 +- .../io/xlang/debezium/debezium_test.go | 2 +- sdks/java/io/debezium/build.gradle | 17 +++++++---- .../debezium/expansion-service/build.gradle | 6 ++-- sdks/java/io/debezium/src/README.md | 6 ++-- .../io/debezium/KafkaSourceConsumerFn.java | 5 ++++ .../debezium/DebeziumIOMySqlConnectorIT.java | 6 ++-- .../DebeziumIOPostgresSqlConnectorIT.java | 4 +-- .../DebeziumReadSchemaTransformTest.java | 29 ++++++++++++++----- .../io/external/xlang_debeziumio_it_test.py | 2 +- 10 files changed, 53 insertions(+), 26 deletions(-) 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/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/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", From 0d0e9551159563b2cd85acdfec0f268f13cc380a Mon Sep 17 00:00:00 2001 From: Tobias Kaymak Date: Wed, 22 Jul 2026 10:12:48 +0000 Subject: [PATCH 3/3] Wire DebeziumSDFDatabaseHistory default to schema.history.internal The default schema history implementation was registered under the pre-Debezium-2.0 key database.history, which Debezium 3.x ignores. As a result DebeziumSDFDatabaseHistory was unreachable unless users set schema.history.internal explicitly and historized connectors silently fell back to KafkaSchemaHistory pointing at localhost:9092. Registering the default under schema.history.internal makes the Beam native history implementation the default again as originally intended. Users who set schema.history.internal themselves are unaffected. --- .../main/java/org/apache/beam/io/debezium/DebeziumIO.java | 8 ++++---- .../java/org/apache/beam/io/debezium/DebeziumIOTest.java | 3 ++- 2 files changed, 6 insertions(+), 5 deletions(-) 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 1f967ba8316a..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 @@ -74,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(); @@ -640,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/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