[DEX-270] Use HZ EE connector for CDC tests (#192)

hazelcast-jet-ansible PR: hazelcast/hazelcast-jet-ansible#141 Jenkins job: https://jenkins.hazelcast.com/view/Jet%20-%20Misc/job/jet-soak-tests-ondrej/290/console - 15hours soak test passed - `isolated-jobs-stream-test.log isolated-jobs-batch-test.log` failed due to the license and it's not related to this PR - it was executed with `.setProperty("database.server.id", clusterName.contains(STABLE_CLUSTER) ? 444444 : 555555)` since there was issue in the connector with `setDatabaseClientId`. However it's already fixed hence `.setDatabaseClientId(clusterName.contains(STABLE_CLUSTER) ? 444444 : 555555)` is used in this PR.
hazelcast · Sep 13, 2024 · f60cfa8 · f60cfa8
1 parent ca3fa00
commit f60cfa8
Show file tree

Hide file tree

Showing 4 changed files with 23 additions and 19 deletions.
diff --git a/cdc-sink-test/pom.xml b/cdc-sink-test/pom.xml
@@ -19,7 +19,7 @@
     <dependencies>
         <dependency>
             <groupId>com.hazelcast.jet</groupId>
-            <artifactId>hazelcast-jet-cdc-debezium</artifactId>
+            <artifactId>hazelcast-enterprise-cdc-debezium</artifactId>
             <version>${hazelcast.version}</version>
         </dependency>
         <dependency>

diff --git a/cdc-sink-test/src/main/java/com/hazelcast/jet/tests/cdc/sink/CdcSinkTest.java b/cdc-sink-test/src/main/java/com/hazelcast/jet/tests/cdc/sink/CdcSinkTest.java
@@ -18,10 +18,10 @@
 
 import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.jet.Job;
-import com.hazelcast.jet.cdc.CdcSinks;
-import com.hazelcast.jet.cdc.ChangeRecord;
-import com.hazelcast.jet.cdc.Operation;
-import com.hazelcast.jet.cdc.impl.ChangeRecordImpl;
+import com.hazelcast.enterprise.jet.cdc.CdcSinks;
+import com.hazelcast.enterprise.jet.cdc.ChangeRecord;
+import com.hazelcast.enterprise.jet.cdc.Operation;
+import com.hazelcast.enterprise.jet.cdc.impl.ChangeRecordImpl;
 import com.hazelcast.jet.config.JobConfig;
 import com.hazelcast.jet.config.ProcessingGuarantee;
 import com.hazelcast.jet.pipeline.Pipeline;
@@ -37,10 +37,10 @@
 import java.util.Set;
 import java.util.function.Supplier;
 
-import static com.hazelcast.jet.cdc.Operation.DELETE;
-import static com.hazelcast.jet.cdc.Operation.INSERT;
-import static com.hazelcast.jet.cdc.Operation.SYNC;
-import static com.hazelcast.jet.cdc.Operation.UPDATE;
+import static com.hazelcast.enterprise.jet.cdc.Operation.DELETE;
+import static com.hazelcast.enterprise.jet.cdc.Operation.INSERT;
+import static com.hazelcast.enterprise.jet.cdc.Operation.SYNC;
+import static com.hazelcast.enterprise.jet.cdc.Operation.UPDATE;
 import static com.hazelcast.jet.core.JobStatus.FAILED;
 import static com.hazelcast.jet.core.JobStatus.RUNNING;
 import static com.hazelcast.jet.tests.common.Util.getJobStatusWithRetry;
@@ -244,6 +244,7 @@ private static ChangeRecord record(int messageId, Operation op, String key, Stri
                 messageId,
                 op,
                 key,
+                () -> "notUsed",
                 oldValue,
                 newValue,
                 "table",

diff --git a/cdc-source-test/pom.xml b/cdc-source-test/pom.xml
@@ -19,12 +19,12 @@
     <dependencies>
         <dependency>
             <groupId>com.hazelcast.jet</groupId>
-            <artifactId>hazelcast-jet-cdc-debezium</artifactId>
+            <artifactId>hazelcast-enterprise-cdc-debezium</artifactId>
             <version>${hazelcast.version}</version>
         </dependency>
         <dependency>
             <groupId>com.hazelcast.jet</groupId>
-            <artifactId>hazelcast-jet-cdc-mysql</artifactId>
+            <artifactId>hazelcast-enterprise-cdc-mysql</artifactId>
             <version>${hazelcast.version}</version>
         </dependency>
         <dependency>

diff --git a/cdc-source-test/src/main/java/com/hazelcast/jet/tests/cdc/source/CdcSourceTest.java b/cdc-source-test/src/main/java/com/hazelcast/jet/tests/cdc/source/CdcSourceTest.java
@@ -18,10 +18,10 @@
 
 import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.jet.Job;
-import com.hazelcast.jet.cdc.ChangeRecord;
-import com.hazelcast.jet.cdc.Operation;
-import com.hazelcast.jet.cdc.RecordPart;
-import com.hazelcast.jet.cdc.mysql.MySqlCdcSources;
+import com.hazelcast.enterprise.jet.cdc.ChangeRecord;
+import com.hazelcast.enterprise.jet.cdc.Operation;
+import com.hazelcast.enterprise.jet.cdc.RecordPart;
+import com.hazelcast.enterprise.jet.cdc.mysql.MySqlCdcSources;
 import com.hazelcast.jet.config.JobConfig;
 import com.hazelcast.jet.config.ProcessingGuarantee;
 import com.hazelcast.jet.pipeline.Pipeline;
@@ -156,10 +156,10 @@ private Pipeline pipeline(String clusterName, String tableName) {
                 .setDatabasePort(3306)
                 .setDatabaseUser("debezium")
                 .setDatabasePassword("Dbz,1234")
-                .setClusterName("dbserver1")
+                .setDatabaseName("cdc_source_test_db")
                 .setDatabaseClientId(clusterName.contains(STABLE_CLUSTER) ? 444444 : 555555)
-                .setDatabaseWhitelist(DATABASE_NAME)
-                .setTableWhitelist(DATABASE_NAME + "." + tableName)
+                .setSchemaIncludeList(DATABASE_NAME)
+                .setTableIncludeList(DATABASE_NAME + "." + tableName)
                 .build();
 
         Sink<Integer> insertSink = Sinks.fromProcessor("insertVerificationProcessor",
@@ -174,7 +174,10 @@ private Pipeline pipeline(String clusterName, String tableName) {
                           .withNativeTimestamps(0)
                           .map(record -> {
                               RecordPart value = record.value();
-                              int id = (int) value.toMap().get("id");
+                              int id = -1;
+                              if (value != null && value.toMap().get("id") != null) {
+                                   id = (int) value.toMap().get("id");
+                              }
                               return entry(record.operation(), id);
                           });