fix cdc java ut

risingwavelabs · Oct 16, 2023 · 014b2ab · 014b2ab
1 parent e03cad0
commit 014b2ab
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 7 deletions.
diff --git a/...ingwave-connector-test/src/test/java/com/risingwave/connector/source/MySQLSourceTest.java b/...ingwave-connector-test/src/test/java/com/risingwave/connector/source/MySQLSourceTest.java
@@ -124,10 +124,12 @@ public void testLines() throws InterruptedException, SQLException {
                     int count = 0;
                     while (eventStream.hasNext()) {
                         List<CdcMessage> messages = eventStream.next().getEventsList();
-                        for (CdcMessage ignored : messages) {
-                            count++;
+                        for (CdcMessage msg : messages) {
+                            if (!msg.getPayload().isBlank()) {
+                                count++;
+                            }
                         }
-                        if (count == 10000) {
+                        if (count >= 10000) {
                             return count;
                         }
                     }

diff --git a/...wave-connector-test/src/test/java/com/risingwave/connector/source/PostgresSourceTest.java b/...wave-connector-test/src/test/java/com/risingwave/connector/source/PostgresSourceTest.java
@@ -130,10 +130,12 @@ public void testLines() throws Exception {
                     while (eventStream.hasNext()) {
                         List<ConnectorServiceProto.CdcMessage> messages =
                                 eventStream.next().getEventsList();
-                        for (ConnectorServiceProto.CdcMessage ignored : messages) {
-                            count++;
+                        for (ConnectorServiceProto.CdcMessage msg : messages) {
+                            if (!msg.getPayload().isBlank()) {
+                                count++;
+                            }
                         }
-                        if (count == 10000) {
+                        if (count >= 10000) {
                             return count;
                         }
                     }

diff --git a/...ngwave-connector-test/src/test/java/com/risingwave/connector/source/SourceTestClient.java b/...ngwave-connector-test/src/test/java/com/risingwave/connector/source/SourceTestClient.java
@@ -41,7 +41,7 @@ public class SourceTestClient {
     static final Logger LOG = LoggerFactory.getLogger(SourceTestClient.class.getName());
 
     // default port for connector service
-    static final int DEFAULT_PORT = 50051;
+    static final int DEFAULT_PORT = 60051;
     private final ConnectorServiceGrpc.ConnectorServiceBlockingStub blockingStub;
 
     public Properties sqlStmts = new Properties();