Upgrade debezium version to 2.5.4.Final

Author: nikhil-ctds

pom.xml

@@ -197,7 +197,7 @@ flexible messaging model and an intuitive client API.</description>
     <opensearch.version>1.2.4</opensearch.version>
     <elasticsearch-java.version>8.12.1</elasticsearch-java.version>
     <trino.version>368</trino.version>
-    <debezium.version>1.9.7.Final</debezium.version>
+    <debezium.version>2.5.4.Final</debezium.version>
     <debezium.postgresql.version>42.5.5</debezium.postgresql.version>
     <debezium.mysql.version>8.0.30</debezium.mysql.version>
     <!-- Override version that brings CVE-2022-3143 with debezium -->

pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/DebeziumSource.java

@@ -88,7 +88,7 @@ public void open(Map<String, Object> config, SourceContext sourceContext) throws
         setConfigIfNull(config, PulsarKafkaWorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, DEFAULT_CONVERTER);
 
         // database.history : implementation class for database history.
-        setConfigIfNull(config, HistorizedRelationalDatabaseConnectorConfig.DATABASE_HISTORY.name(), DEFAULT_HISTORY);
+        setConfigIfNull(config, HistorizedRelationalDatabaseConnectorConfig.SCHEMA_HISTORY.name(), DEFAULT_HISTORY);
 
         // database.history.pulsar.service.url
         String pulsarUrl = (String) config.get(PulsarDatabaseHistory.SERVICE_URL.name());

pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistory.java

@@ -26,12 +26,12 @@
 import io.debezium.config.Configuration;
 import io.debezium.config.Field;
 import io.debezium.document.DocumentReader;
-import io.debezium.relational.history.AbstractDatabaseHistory;
-import io.debezium.relational.history.DatabaseHistory;
-import io.debezium.relational.history.DatabaseHistoryException;
-import io.debezium.relational.history.DatabaseHistoryListener;
+import io.debezium.relational.history.AbstractSchemaHistory;
 import io.debezium.relational.history.HistoryRecord;
 import io.debezium.relational.history.HistoryRecordComparator;
+import io.debezium.relational.history.SchemaHistory;
+import io.debezium.relational.history.SchemaHistoryException;
+import io.debezium.relational.history.SchemaHistoryListener;
 import java.io.IOException;
 import java.util.Collections;
 import java.util.HashMap;
@@ -52,12 +52,12 @@
 import org.apache.pulsar.client.api.Schema;
 
 /**
- * A {@link DatabaseHistory} implementation that records schema changes as normal pulsar messages on the specified
+ * A {@link SchemaHistory} implementation that records schema changes as normal pulsar messages on the specified
  * topic, and that recovers the history by establishing a Kafka Consumer re-processing all messages on that topic.
  */
 @Slf4j
 @ThreadSafe
-public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
+public final class PulsarDatabaseHistory extends AbstractSchemaHistory {
 
     public static final Field TOPIC = Field.create(CONFIGURATION_FIELD_PREFIX_STRING + "pulsar.topic")
         .withDisplayName("Database history topic name")
@@ -97,7 +97,7 @@ public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
         TOPIC,
         SERVICE_URL,
         CLIENT_BUILDER,
-        DatabaseHistory.NAME,
+        SchemaHistory.NAME,
         READER_CONFIG);
 
     private final ObjectMapper mapper = new ObjectMapper();
@@ -113,7 +113,7 @@ public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
     public void configure(
             Configuration config,
             HistoryRecordComparator comparator,
-            DatabaseHistoryListener listener,
+            SchemaHistoryListener listener,
             boolean useCatalogBeforeSchema) {
         super.configure(config, comparator, listener, useCatalogBeforeSchema);
         if (!config.validateAndRecord(ALL_FIELDS, logger::error)) {
@@ -148,7 +148,7 @@ public void configure(
         }
 
         // Copy the relevant portions of the configuration and add useful defaults ...
-        this.dbHistoryName = config.getString(DatabaseHistory.NAME, UUID.randomUUID().toString());
+        this.dbHistoryName = config.getString(SchemaHistory.NAME, UUID.randomUUID().toString());
 
         log.info("Configure to store the debezium database history {} to pulsar topic {}",
             dbHistoryName, topicName);
@@ -201,7 +201,7 @@ public void start() {
     }
 
     @Override
-    protected void storeRecord(HistoryRecord record) throws DatabaseHistoryException {
+    protected void storeRecord(HistoryRecord record) throws SchemaHistoryException {
         if (this.producer == null) {
             throw new IllegalStateException("No producer is available. Ensure that 'start()'"
                     + " is called before storing database history records.");
@@ -212,7 +212,7 @@ protected void storeRecord(HistoryRecord record) throws DatabaseHistoryException
         try {
             producer.send(record.toString());
         } catch (PulsarClientException e) {
-            throw new DatabaseHistoryException(e);
+            throw new SchemaHistoryException(e);
         }
     }
 

pulsar-io/debezium/core/src/test/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistoryTest.java

@@ -27,8 +27,8 @@
 import io.debezium.connector.mysql.antlr.MySqlAntlrDdlParser;
 import io.debezium.relational.Tables;
 import io.debezium.relational.ddl.DdlParser;
-import io.debezium.relational.history.DatabaseHistory;
-import io.debezium.relational.history.DatabaseHistoryListener;
+import io.debezium.relational.history.SchemaHistory;
+import io.debezium.relational.history.SchemaHistoryListener;
 import io.debezium.text.ParsingException;
 import io.debezium.util.Collect;
 
@@ -80,8 +80,8 @@ protected void cleanup() throws Exception {
     private void testHistoryTopicContent(boolean skipUnparseableDDL, boolean testWithClientBuilder, boolean testWithReaderConfig) throws Exception {
         Configuration.Builder configBuidler = Configuration.create()
                 .with(PulsarDatabaseHistory.TOPIC, topicName)
-                .with(DatabaseHistory.NAME, "my-db-history")
-                .with(DatabaseHistory.SKIP_UNPARSEABLE_DDL_STATEMENTS, skipUnparseableDDL);
+                .with(SchemaHistory.NAME, "my-db-history")
+                .with(SchemaHistory.SKIP_UNPARSEABLE_DDL_STATEMENTS, skipUnparseableDDL);
 
         if (testWithClientBuilder) {
             ClientBuilder builder = PulsarClient.builder().serviceUrl(brokerUrl.toString());
@@ -101,7 +101,7 @@ private void testHistoryTopicContent(boolean skipUnparseableDDL, boolean testWit
         }
 
         // Start up the history ...
-        history.configure(configBuidler.build(), null, DatabaseHistoryListener.NOOP, true);
+        history.configure(configBuidler.build(), null, SchemaHistoryListener.NOOP, true);
         history.start();
 
         // Should be able to call start more than once ...
@@ -160,7 +160,7 @@ private void testHistoryTopicContent(boolean skipUnparseableDDL, boolean testWit
         // Stop the history (which should stop the producer) ...
         history.stop();
         history = new PulsarDatabaseHistory();
-        history.configure(configBuidler.build(), null, DatabaseHistoryListener.NOOP, true);
+        history.configure(configBuidler.build(), null, SchemaHistoryListener.NOOP, true);
         // no need to start
 
         // Recover from the very beginning to just past the first change ...
@@ -240,11 +240,11 @@ public void testExists() throws Exception {
         Configuration config = Configuration.create()
             .with(PulsarDatabaseHistory.SERVICE_URL, brokerUrl.toString())
             .with(PulsarDatabaseHistory.TOPIC, "persistent://my-property/my-ns/dummytopic")
-            .with(DatabaseHistory.NAME, "my-db-history")
-            .with(DatabaseHistory.SKIP_UNPARSEABLE_DDL_STATEMENTS, true)
+            .with(SchemaHistory.NAME, "my-db-history")
+            .with(SchemaHistory.SKIP_UNPARSEABLE_DDL_STATEMENTS, true)
             .build();
 
-        history.configure(config, null, DatabaseHistoryListener.NOOP, true);
+        history.configure(config, null, SchemaHistoryListener.NOOP, true);
         history.start();
 
         // dummytopic should not exist yet

tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumOracleDbSourceTester.java

@@ -54,16 +54,19 @@ public DebeziumOracleDbSourceTester(PulsarCluster cluster) {
 
         pulsarServiceUrl = "pulsar://pulsar-proxy:" + PulsarContainer.BROKER_PORT;
 
+        sourceConfig.put("connector.class", "io.debezium.connector.oracle.OracleConnector");
         sourceConfig.put("database.hostname", DebeziumOracleDbContainer.NAME);
         sourceConfig.put("database.port", "1521");
         sourceConfig.put("database.user", "dbzuser");
         sourceConfig.put("database.password", "dbz");
         sourceConfig.put("database.server.name", "XE");
         sourceConfig.put("database.dbname", "XE");
+        sourceConfig.put("topic.prefix", "topic");
         sourceConfig.put("snapshot.mode", "schema_only");
 
         sourceConfig.put("schema.include.list", "inv");
-        sourceConfig.put("database.history.pulsar.service.url", pulsarServiceUrl);
+        sourceConfig.put("schema.history.internal", "org.apache.pulsar.io.debezium.oracle.PulsarDatabaseHistory");
+        sourceConfig.put("schema.history.internal.pulsar.service.url", pulsarServiceUrl);
         sourceConfig.put("topic.namespace", "debezium/oracle");
     }