Wait for VStream started in startConnector

debezium · Dec 2, 2021 · f0c3364 · f0c3364
1 parent cd62b46
commit f0c3364
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 2 deletions.
diff --git a/src/main/java/io/debezium/connector/vitess/connection/VitessReplicationConnection.java b/src/main/java/io/debezium/connector/vitess/connection/VitessReplicationConnection.java
@@ -171,6 +171,7 @@ private int getNumOfRowEvents(Vtgate.VStreamResponse response) {
                         .setFlags(vStreamFlags)
                         .build(),
                 responseObserver);
+        LOGGER.info("Started VStream");
     }
 
     private VitessGrpc.VitessStub newStub(ManagedChannel channel) {

diff --git a/src/test/java/io/debezium/connector/vitess/AbstractVitessConnectorTest.java b/src/test/java/io/debezium/connector/vitess/AbstractVitessConnectorTest.java
@@ -202,8 +202,6 @@ protected static String incrementGtid(String gtid, int increment) {
 
     protected TestConsumer testConsumer(int expectedRecordsCount, String... topicPrefixes) throws InterruptedException {
         TestConsumer consumer = new TestConsumer(expectedRecordsCount, topicPrefixes);
-        // Sleep for 1 second before return to avoid race conditions.
-        Thread.sleep(1000);
         return consumer;
     }
 

diff --git a/src/test/java/io/debezium/connector/vitess/VitessConnectorIT.java b/src/test/java/io/debezium/connector/vitess/VitessConnectorIT.java
@@ -559,6 +559,13 @@ private void waitForShardedGtidAcquiring(final LogInterceptor logInterceptor) {
                 .until(() -> logInterceptor.containsMessage("Default VGTID '[{\"keyspace\":"));
     }
 
+    private void waitForVStreamStarted(final LogInterceptor logInterceptor) {
+        // The inserts must happen only after VStream is started some buffer.
+        Awaitility.await().atMost(Duration.ofSeconds(TestHelper.waitTimeForRecords()))
+                .pollInterval(Duration.ofMillis(500))
+                .until(() -> logInterceptor.containsMessage("Started VStream"));
+    }
+
     private void startConnector() throws InterruptedException {
         startConnector(false);
     }
@@ -576,9 +583,11 @@ private void startConnector(boolean hasMultipleShards) throws InterruptedExcepti
     private void startConnector(Function<Configuration.Builder, Configuration.Builder> customConfig, boolean hasMultipleShards)
             throws InterruptedException {
         Configuration.Builder configBuilder = customConfig.apply(TestHelper.defaultConfig(hasMultipleShards));
+        final LogInterceptor logInterceptor = new LogInterceptor();
         start(VitessConnector.class, configBuilder.build());
         assertConnectorIsRunning();
         waitForStreamingRunning();
+        waitForVStreamStarted(logInterceptor);
     }
 
     private void waitForStreamingRunning() throws InterruptedException {