Fix flaky test for debezium (apache#5878)

instructure · Dec 17, 2019 · fe0ddcc · fe0ddcc
1 parent 9b55324
commit fe0ddcc
Showing 1 changed file with 40 additions and 0 deletions.
diff --git a/...tion/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java b/...tion/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
@@ -2073,6 +2073,16 @@ private  void testDebeziumMySqlConnect()
 
         @Cleanup
         PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(pulsarCluster.getHttpServiceUrl()).build();
+        try {
+            // If topic already exists, we should delete it so as not to affect the following tests.
+            admin.topics().getStats(consumeTopicName);
+            admin.topics().delete(consumeTopicName);
+            admin.schemas().deleteSchema(consumeTopicName);
+        } catch (PulsarAdminException e) {
+            // Expected results, ignoring the exception
+            log.info("Topic: {} does not exist, we can continue the following tests. Exceptions message: {}",
+                    consumeTopicName, e.getMessage());
+        }
         admin.topics().createNonPartitionedTopic(consumeTopicName);
         admin.topics().createNonPartitionedTopic(outputTopicName);
 
@@ -2133,6 +2143,21 @@ private  void testDebeziumPostgreSqlConnect() throws Exception {
                 .serviceUrl(pulsarCluster.getPlainTextServiceUrl())
                 .build();
 
+        @Cleanup
+        PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(pulsarCluster.getHttpServiceUrl()).build();
+        try {
+            // If topic already exists, we should delete it so as not to affect the following tests.
+            admin.topics().getStats(consumeTopicName);
+            admin.topics().delete(consumeTopicName);
+            admin.schemas().deleteSchema(consumeTopicName);
+        } catch (PulsarAdminException e) {
+            // Expected results, ignoring the exception
+            log.info("Topic: {} does not exist, we can continue the following tests. Exceptions message: {}",
+                    consumeTopicName, e.getMessage());
+        }
+        admin.topics().createNonPartitionedTopic(consumeTopicName);
+        admin.topics().createNonPartitionedTopic(outputTopicName);
+
         @Cleanup
         Consumer<KeyValue<byte[], byte[]>> consumer = client.newConsumer(KeyValueSchema.kvBytes())
                 .topic(consumeTopicName)
@@ -2190,6 +2215,21 @@ private  void testDebeziumMongoDbConnect() throws Exception {
                 .serviceUrl(pulsarCluster.getPlainTextServiceUrl())
                 .build();
 
+        @Cleanup
+        PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(pulsarCluster.getHttpServiceUrl()).build();
+        try {
+            // If topic already exists, we should delete it so as not to affect the following tests.
+            admin.topics().getStats(consumeTopicName);
+            admin.topics().delete(consumeTopicName);
+            admin.schemas().deleteSchema(consumeTopicName);
+        } catch (PulsarAdminException e) {
+            // Expected results, ignoring the exception
+            log.info("Topic: {} does not exist, we can continue the following tests. Exceptions message: {}",
+                    consumeTopicName, e.getMessage());
+        }
+        admin.topics().createNonPartitionedTopic(consumeTopicName);
+        admin.topics().createNonPartitionedTopic(outputTopicName);
+
         @Cleanup
         Consumer<KeyValue<byte[], byte[]>> consumer = client.newConsumer(KeyValueSchema.kvBytes())
                 .topic(consumeTopicName)