Fix flaky test for debezium
@@ -2073,6 +2073,16 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
@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 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
.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 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
.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)
