Updated configs

datastax · Jul 22, 2024 · 0656cdb · 0656cdb
1 parent ee5a4ac
commit 0656cdb
Show file tree

Hide file tree

Showing 7 changed files with 28 additions and 34 deletions.
diff --git a/...rc/test/java/org/apache/pulsar/tests/integration/containers/DebeziumMongoDbContainer.java b/...rc/test/java/org/apache/pulsar/tests/integration/containers/DebeziumMongoDbContainer.java
@@ -25,7 +25,7 @@ public class DebeziumMongoDbContainer extends ChaosContainer<DebeziumMongoDbCont
     public static final String NAME = "debezium-mongodb-example";
 
     public static final Integer[] PORTS = { 27017 };
-    private static final String IMAGE_NAME = "debezium/example-mongodb:0.10";
+    private static final String IMAGE_NAME = "debezium/example-mongodb:2.5.0.Final";
 
     public DebeziumMongoDbContainer(String clusterName) {
         super(clusterName, IMAGE_NAME);

diff --git a/.../org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMongoDbSourceTester.java b/.../org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMongoDbSourceTester.java
@@ -44,19 +44,14 @@ public DebeziumMongoDbSourceTester(PulsarCluster cluster) {
         this.pulsarCluster = cluster;
         pulsarServiceUrl = "pulsar://pulsar-proxy:" + PulsarContainer.BROKER_PORT;
 
-        sourceConfig.put("mongodb.hosts", "rs0/" + DebeziumMongoDbContainer.NAME + ":27017");
-        sourceConfig.put("mongodb.name", "dbserver1");
+        sourceConfig.put("connector.class", "io.debezium.connector.mongodb.MongoDbConnector");
+        sourceConfig.put("mongodb.connection.string", "mongodb://" + DebeziumMongoDbContainer.NAME + ":27017/?replicaSet=rs0");
         sourceConfig.put("mongodb.user", "debezium");
         sourceConfig.put("mongodb.password", "dbz");
-        sourceConfig.put("mongodb.task.id","1");
-        sourceConfig.put("database.include.list", "inventory");
+        sourceConfig.put("mongodb.task.id", "1");
         sourceConfig.put("schema.history.internal.pulsar.service.url", pulsarServiceUrl);
         sourceConfig.put("topic.namespace", "debezium/mongodb");
-        sourceConfig.put("capture.mode", "oplog");
-        sourceConfig.put("connector.class", "io.debezium.connector.mongodb.MongoDbConnector");
-        sourceConfig.put("topic.prefix", "test");
-        sourceConfig.put("collection.include.list", "inventory");
-        sourceConfig.put("mongodb.connection.string", "mongodb://" + DebeziumMongoDbContainer.NAME + ":27017/?replicaSet=rs0");
+        sourceConfig.put("topic.prefix", "dbserver1");
     }
 
     @Override
@@ -75,10 +70,10 @@ public void prepareSource() throws Exception {
     @Override
     public void prepareInsertEvent() throws Exception {
         this.debeziumMongoDbContainer.execCmd("/bin/bash", "-c",
-                "mongo -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory " +
+                "mongosh -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory " +
                         "--eval 'db.products.find()'");
         this.debeziumMongoDbContainer.execCmd("/bin/bash", "-c",
-                "mongo -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory " +
+                "mongosh -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory " +
                         "--eval 'db.products.insert({ " +
                         "_id : NumberLong(\"110\")," +
                         "name : \"test-debezium\"," +
@@ -90,20 +85,20 @@ public void prepareInsertEvent() throws Exception {
     @Override
     public void prepareDeleteEvent() throws Exception {
         this.debeziumMongoDbContainer.execCmd("/bin/bash", "-c",
-                "mongo -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory " +
+                "mongosh -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory " +
                         "--eval 'db.products.find()'");
         this.debeziumMongoDbContainer.execCmd("/bin/bash", "-c",
-                "mongo -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory " +
+                "mongosh -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory " +
                         "--eval 'db.products.deleteOne({name : \"test-debezium-update\"})'");
     }
 
     @Override
     public void prepareUpdateEvent() throws Exception {
         this.debeziumMongoDbContainer.execCmd("/bin/bash", "-c",
-                "mongo -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory " +
+                "mongosh -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory " +
                         "--eval 'db.products.find()'");
         this.debeziumMongoDbContainer.execCmd("/bin/bash", "-c",
-                "mongo -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory " +
+                "mongosh -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory " +
                         "--eval 'db.products.update({" +
                         "_id : 110}," +
                         "{$set:{name:\"test-debezium-update\", description: \"this is update description\"}})'");

diff --git a/...va/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMsSqlSourceTester.java b/...va/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMsSqlSourceTester.java
@@ -56,19 +56,18 @@ public DebeziumMsSqlSourceTester(PulsarCluster cluster) {
 
         pulsarServiceUrl = "pulsar://pulsar-proxy:" + PulsarContainer.BROKER_PORT;
 
+        sourceConfig.put("connector.class", "io.debezium.connector.sqlserver.SqlServerConnector");
         sourceConfig.put("database.hostname", DebeziumMsSqlContainer.NAME);
         sourceConfig.put("database.port", "1433");
         sourceConfig.put("database.user", "sa");
         sourceConfig.put("database.password", DebeziumMsSqlContainer.SA_PASSWORD);
         sourceConfig.put("database.server.name", "mssql");
-        sourceConfig.put("database.dbname", "TestDB");
-        sourceConfig.put("snapshot.mode", "schema_only");
+        sourceConfig.put("database.names", "TestDB");
         sourceConfig.put("schema.history.internal.pulsar.service.url", pulsarServiceUrl);
         sourceConfig.put("topic.namespace", "debezium/mssql");
-        sourceConfig.put("connector.class", "io.debezium.connector.sqlserver.SqlServerConnector");
-        sourceConfig.put("topic.prefix", "test");
-        sourceConfig.put("database.names", "TestDB");
-        sourceConfig.put("table.include.list", "inventory");
+        sourceConfig.put("topic.prefix", "mssql");
+        sourceConfig.put("database.encrypt", "false");
+        sourceConfig.put("task.id", "1");
     }
 
     @Override

diff --git a/...va/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMySqlSourceTester.java b/...va/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMySqlSourceTester.java
@@ -55,24 +55,22 @@ public DebeziumMySqlSourceTester(PulsarCluster cluster, String converterClassNam
         this.pulsarCluster = cluster;
         pulsarServiceUrl = "pulsar://pulsar-proxy:" + PulsarContainer.BROKER_PORT;
 
+        sourceConfig.put("connector.class", "io.debezium.connector.mysql.MySqlConnector");
         sourceConfig.put("database.hostname", DebeziumMySQLContainer.NAME);
         sourceConfig.put("database.port", "3306");
         sourceConfig.put("database.user", "debezium");
         sourceConfig.put("database.password", "dbz");
         sourceConfig.put("database.server.id", "184054");
         sourceConfig.put("database.server.name", "dbserver1");
-        sourceConfig.put("database.whitelist", "inventory");
+        sourceConfig.put("database.include.list", "inventory");
         if (!testWithClientBuilder) {
             sourceConfig.put("schema.history.internal.pulsar.service.url", pulsarServiceUrl);
         }
         sourceConfig.put("key.converter", converterClassName);
         sourceConfig.put("value.converter", converterClassName);
         sourceConfig.put("topic.namespace", "debezium/mysql-" +
                 (converterClassName.endsWith("AvroConverter") ? "avro" : "json"));
-        sourceConfig.put("connector.class", "io.debezium.connector.mysql.MySqlConnector");
-        sourceConfig.put("topic.prefix", "test");
-        sourceConfig.put("database.include.list", "inventory");
-        sourceConfig.put("include.schema.changes", "true");
+        sourceConfig.put("topic.prefix", "dbserver1");
     }
 
     @Override

diff --git a/...org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumOracleDbSourceTester.java b/...org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumOracleDbSourceTester.java
@@ -56,18 +56,18 @@ 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("snapshot.mode", "schema_only");
+        sourceConfig.put("database.whitelist", "inv");
         sourceConfig.put("schema.include.list", "inv");
         sourceConfig.put("schema.history.internal.pulsar.service.url", pulsarServiceUrl);
         sourceConfig.put("topic.namespace", "debezium/oracle");
-        sourceConfig.put("connector.class", "io.debezium.connector.oracle.OracleConnector");
-        sourceConfig.put("topic.prefix", "test");
+        sourceConfig.put("topic.prefix", "XE");
     }
 
     @Override

diff --git a/...g/apache/pulsar/tests/integration/io/sources/debezium/DebeziumPostgreSqlSourceTester.java b/...g/apache/pulsar/tests/integration/io/sources/debezium/DebeziumPostgreSqlSourceTester.java
@@ -67,6 +67,7 @@ public DebeziumPostgreSqlSourceTester(PulsarCluster cluster) {
 
         pulsarServiceUrl = "pulsar://pulsar-proxy:" + PulsarContainer.BROKER_PORT;
 
+        sourceConfig.put("connector.class", "io.debezium.connector.postgresql.PostgresConnector");
         sourceConfig.put("database.hostname", DebeziumPostgreSqlContainer.NAME);
         sourceConfig.put("database.port", "5432");
         sourceConfig.put("database.user", "postgres");
@@ -75,11 +76,12 @@ public DebeziumPostgreSqlSourceTester(PulsarCluster cluster) {
         sourceConfig.put("database.server.name", "dbserver1");
         sourceConfig.put("database.dbname", "postgres");
         sourceConfig.put("schema.whitelist", "inventory");
-        sourceConfig.put("table.blacklist", "inventory.spatial_ref_sys,inventory.geom");
+        sourceConfig.put("plugin.name", "pgoutput");
         sourceConfig.put("schema.history.internal.pulsar.service.url", pulsarServiceUrl);
+        sourceConfig.put("schema.history.internal", "org.apache.pulsar.io.debezium.PulsarDatabaseHistory");
         sourceConfig.put("topic.namespace", "debezium/postgresql");
-        sourceConfig.put("connector.class", "io.debezium.connector.postgresql.PostgresConnector");
-        sourceConfig.put("topic.prefix", "test");
+        sourceConfig.put("topic.prefix", "dbserver1");
+        sourceConfig.put("table.blacklist", "inventory.spatial_ref_sys,inventory.geom");
         sourceConfig.put("table.include.list", "inventory");
     }
 

diff --git a/...va/org/apache/pulsar/tests/integration/io/sources/debezium/PulsarDebeziumSourcesTest.java b/...va/org/apache/pulsar/tests/integration/io/sources/debezium/PulsarDebeziumSourcesTest.java
@@ -211,7 +211,7 @@ private void testDebeziumMsSqlConnect(String converterClassName, boolean jsonWit
         final String tenant = TopicName.PUBLIC_TENANT;
         final String namespace = TopicName.DEFAULT_NAMESPACE;
         final String outputTopicName = "debe-output-topic-name-" + testId.getAndIncrement();
-        final String consumeTopicName = "debezium/mssql/mssql.dbo.customers";
+        final String consumeTopicName = "debezium/mssql/mssql.TestDB.dbo.customers";
         final String sourceName = "test-source-debezium-mssql-" + functionRuntimeType + "-" + randomName(8);
 
         final int numMessages = 1;