You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/01/16 12:25:04 UTC
(pulsar) 05/07: [fix][test]Fix flaky test testTopicUnloadAfterSessionRebuild (#23852)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d93750223f6547bd0470e9602226eb7919f004c0
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Thu Jan 16 16:40:31 2025 +0800
[fix][test]Fix flaky test testTopicUnloadAfterSessionRebuild (#23852)
(cherry picked from commit b3641f068a862a738aacad14b0534cf012e4e584)
---
.../pulsar/broker/service/ZkSessionExpireTest.java | 21 ++++++++++++++++++---
1 file changed, 18 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ZkSessionExpireTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ZkSessionExpireTest.java
index dd0d7c423b6..b05dae2c3c2 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ZkSessionExpireTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ZkSessionExpireTest.java
@@ -26,6 +26,7 @@ import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.ServiceConfiguration;
@@ -39,7 +40,7 @@ import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@Slf4j
-@Test(groups = "flaky")
+@Test(groups = "broker")
public class ZkSessionExpireTest extends NetworkErrorTestBase {
private java.util.function.Consumer<ServiceConfiguration> settings;
@@ -94,7 +95,7 @@ public class ZkSessionExpireTest extends NetworkErrorTestBase {
admin2.namespaces().unload(defaultNamespace);
// Confirm all brokers registered.
- Awaitility.await().untilAsserted(() -> {
+ Awaitility.await().atMost(Duration.ofSeconds(20)).untilAsserted(() -> {
assertEquals(getAvailableBrokers(pulsar1).size(), 2);
assertEquals(getAvailableBrokers(pulsar2).size(), 2);
});
@@ -160,7 +161,21 @@ public class ZkSessionExpireTest extends NetworkErrorTestBase {
// Verify: the topic on broker-2 is fine.
Awaitility.await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> {
CompletableFuture<Optional<Topic>> future = pulsar1.getBrokerService().getTopic(topicName, false);
- assertTrue(future == null || future.isCompletedExceptionally());
+ log.info("broker 1 topics {}", pulsar1.getBrokerService().getTopics().keySet());
+ log.info("broker 2 topics {}", pulsar2.getBrokerService().getTopics().keySet());
+ log.info("broker 1 bundles {}", pulsar1.getNamespaceService().getOwnershipCache().getOwnedBundles()
+ .keySet().stream().map(k -> k.getNamespaceObject().toString() + "/" + k.getBundleRange())
+ .filter(s -> s.contains(defaultNamespace)).collect(Collectors.toList()));
+ log.info("broker 2 bundles {}", pulsar2.getNamespaceService().getOwnershipCache().getOwnedBundles()
+ .keySet().stream().map(k -> k.getNamespaceObject().toString() + "/" + k.getBundleRange())
+ .filter(s -> s.contains(defaultNamespace)).collect(Collectors.toList()));
+ log.info("future: {}, isDone: {}, isCompletedExceptionally: {}",
+ future, future == null ? "null" : future.isDone(),
+ future, future == null ? "null" : future.isCompletedExceptionally());
+ assertTrue(future == null
+ || !pulsar1.getBrokerService().getTopics().containsKey(topicName)
+ || (future.isDone() && !future.isCompletedExceptionally() && future.get().isEmpty())
+ || future.isCompletedExceptionally());
});
Topic broker2Topic3 = pulsar2.getBrokerService().getTopic(topicName, false).join().get();
assertNotNull(broker2Topic3);