From 8da1a3548c5a1d4a18b17c566b050f8a1c629cd5 Mon Sep 17 00:00:00 2001 From: penghui Date: Tue, 5 Jul 2022 00:35:44 +0800 Subject: [PATCH 1/3] [fix][flaky-test] Fix failed test NonPersistentTopicE2ETest.testGCWillDeleteSchema --- .../admin/v3/PackagesApiNotEnabledTest.java | 2 +- .../service/NonPersistentTopicE2ETest.java | 51 ++++++++++--------- ...sistentDispatcherFailoverConsumerTest.java | 2 +- 3 files changed, 28 insertions(+), 27 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/PackagesApiNotEnabledTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/PackagesApiNotEnabledTest.java index 3fd39d30d2dd8..d957b9dc2ea54 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/PackagesApiNotEnabledTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/PackagesApiNotEnabledTest.java @@ -44,7 +44,7 @@ protected void cleanup() throws Exception { super.internalCleanup(); } - @Test(timeOut = 60000) + @Test(timeOut = 60000, invocationCount = 1000) public void testPackagesOperationsWithoutPackagesServiceEnabled() { // download package api should return 503 Service Unavailable exception String unknownPackageName = "function://public/default/unknown@v1"; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/NonPersistentTopicE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/NonPersistentTopicE2ETest.java index 0e598f22f1de7..fb1dbfea3f1d7 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/NonPersistentTopicE2ETest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/NonPersistentTopicE2ETest.java @@ -83,7 +83,7 @@ private boolean topicHasSchema(String topicName) { @Test(groups = "broker") public void testGCWillDeleteSchema() throws Exception { // 1. Simple successful GC - String topicName = "non-persistent://prop/ns-abc/topic-1"; + final String topicName = "non-persistent://prop/ns-abc/topic-1"; Producer producer = pulsarClient.newProducer().topic(topicName).create(); producer.close(); @@ -100,56 +100,57 @@ public void testGCWillDeleteSchema() throws Exception { assertTrue(topicHasSchema(topicName)); runGC(); - topic = getTopic(topicName); - assertFalse(topic.isPresent()); + Awaitility.await().untilAsserted(() -> { + assertFalse(getTopic(topicName).isPresent()); + }); assertFalse(topicHasSchema(topicName)); // 1a. Topic that add/removes subscription can be GC'd - topicName = "non-persistent://prop/ns-abc/topic-1a"; + final String topicName2 = "non-persistent://prop/ns-abc/topic-1a"; String subName = "sub1"; - Consumer consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName).subscribe(); - topic = getTopic(topicName); + Consumer consumer = pulsarClient.newConsumer().topic(topicName2).subscriptionName(subName).subscribe(); + topic = getTopic(topicName2); assertTrue(topic.isPresent()); topic.get().addSchema(schemaData).join(); - assertTrue(topicHasSchema(topicName)); + assertTrue(topicHasSchema(topicName2)); - admin.topics().deleteSubscription(topicName, subName); + admin.topics().deleteSubscription(topicName2, subName); consumer.close(); runGC(); - topic = getTopic(topicName); - assertFalse(topic.isPresent()); - assertFalse(topicHasSchema(topicName)); + Awaitility.await().untilAsserted(() -> { + assertFalse(getTopic(topicName2).isPresent()); + }); + assertFalse(topicHasSchema(topicName2)); // 2. Topic is not GCed with live connection - topicName = "non-persistent://prop/ns-abc/topic-2"; + final String topicName3 = "non-persistent://prop/ns-abc/topic-2"; subName = "sub1"; - consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName).subscribe(); - topic = getTopic(topicName); + consumer = pulsarClient.newConsumer().topic(topicName3).subscriptionName(subName).subscribe(); + topic = getTopic(topicName3); assertTrue(topic.isPresent()); topic.get().addSchema(schemaData).join(); - assertTrue(topicHasSchema(topicName)); + assertTrue(topicHasSchema(topicName3)); runGC(); - topic = getTopic(topicName); - assertTrue(topic.isPresent()); - assertTrue(topicHasSchema(topicName)); + assertTrue(getTopic(topicName3).isPresent()); + assertTrue(topicHasSchema(topicName3)); // 3. Topic with subscription is not GCed even with no connections consumer.close(); runGC(); - topic = getTopic(topicName); - assertTrue(topic.isPresent()); - assertTrue(topicHasSchema(topicName)); + assertTrue(getTopic(topicName3).isPresent()); + assertTrue(topicHasSchema(topicName3)); // 4. Topic can be GCed after unsubscribe - admin.topics().deleteSubscription(topicName, subName); + admin.topics().deleteSubscription(topicName3, subName); runGC(); - topic = getTopic(topicName); - assertFalse(topic.isPresent()); - assertFalse(topicHasSchema(topicName)); + Awaitility.await().untilAsserted(() -> { + assertFalse(getTopic(topicName3).isPresent()); + }); + assertFalse(topicHasSchema(topicName3)); } @Test(groups = "broker") diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java index 1a77024a4badc..7d72f431fc9f8 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java @@ -321,7 +321,7 @@ public void testConsumerGroupChangesWithOldNewConsumers() throws Exception { verify(channelCtx, times(1)).writeAndFlush(any(), any()); } - @Test + @Test(invocationCount = 1000) public void testAddRemoveConsumer() throws Exception { log.info("--- Starting PersistentDispatcherFailoverConsumerTest::testAddConsumer ---"); From 8609123cc22e151d83a93c8cf1a0475ec79c187f Mon Sep 17 00:00:00 2001 From: penghui Date: Tue, 5 Jul 2022 08:44:49 +0800 Subject: [PATCH 2/3] Address comments. --- .../pulsar/broker/admin/v3/PackagesApiNotEnabledTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/PackagesApiNotEnabledTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/PackagesApiNotEnabledTest.java index d957b9dc2ea54..3fd39d30d2dd8 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/PackagesApiNotEnabledTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/PackagesApiNotEnabledTest.java @@ -44,7 +44,7 @@ protected void cleanup() throws Exception { super.internalCleanup(); } - @Test(timeOut = 60000, invocationCount = 1000) + @Test(timeOut = 60000) public void testPackagesOperationsWithoutPackagesServiceEnabled() { // download package api should return 503 Service Unavailable exception String unknownPackageName = "function://public/default/unknown@v1"; From 8482f715aea12770a30742ba762151ab173234f7 Mon Sep 17 00:00:00 2001 From: penghui Date: Tue, 5 Jul 2022 08:45:37 +0800 Subject: [PATCH 3/3] Address comments. --- .../service/PersistentDispatcherFailoverConsumerTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java index 7d72f431fc9f8..1a77024a4badc 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java @@ -321,7 +321,7 @@ public void testConsumerGroupChangesWithOldNewConsumers() throws Exception { verify(channelCtx, times(1)).writeAndFlush(any(), any()); } - @Test(invocationCount = 1000) + @Test public void testAddRemoveConsumer() throws Exception { log.info("--- Starting PersistentDispatcherFailoverConsumerTest::testAddConsumer ---");