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 2026/04/22 08:03:26 UTC
(pulsar) 02/02: [fix][test] Fix flaky OneWayReplicatorUsingGlobalZKTest.cleanup (#25389)
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 cb33faaed4292594c9871e4561f22f07947fbdb9
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Mar 23 17:10:08 2026 +0200
[fix][test] Fix flaky OneWayReplicatorUsingGlobalZKTest.cleanup (#25389)
Co-authored-by: Claude Opus 4.6 (1M context) <no...@anthropic.com>
(cherry picked from commit ce3429ce878fd2003793b030b3f317deb49a34db)
---
.../broker/service/OneWayReplicatorTestBase.java | 27 +++++++++++++++++++---
1 file changed, 24 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java
index a2cf9d61499..0b033ec269c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTestBase.java
@@ -303,8 +303,12 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
// delete namespaces.
waitChangeEventsInit(replicatedNamespace);
admin1.namespaces().setNamespaceReplicationClusters(replicatedNamespace, Sets.newHashSet(cluster1));
+ admin1.namespaces().setNamespaceReplicationClusters(
+ sourceClusterAlwaysSchemaCompatibleNamespace, Sets.newHashSet(cluster1));
if (!usingGlobalZK) {
admin2.namespaces().setNamespaceReplicationClusters(replicatedNamespace, Sets.newHashSet(cluster2));
+ admin2.namespaces().setNamespaceReplicationClusters(
+ sourceClusterAlwaysSchemaCompatibleNamespace, Sets.newHashSet(cluster2));
}
// When using global ZK, reducing replication clusters triggers async topic cleanup on removed clusters.
// Retry namespace deletion to handle topics that may be in a transitional state.
@@ -314,16 +318,33 @@ public abstract class OneWayReplicatorTestBase extends TestRetrySupport {
Awaitility.await().atMost(Duration.ofSeconds(30)).ignoreExceptions().untilAsserted(() -> {
admin1.namespaces().deleteNamespace(nonReplicatedNamespace, true);
});
+ Awaitility.await().atMost(Duration.ofSeconds(30)).ignoreExceptions().untilAsserted(() -> {
+ admin1.namespaces().deleteNamespace(sourceClusterAlwaysSchemaCompatibleNamespace, true);
+ });
if (!usingGlobalZK) {
- admin2.namespaces().deleteNamespace(replicatedNamespace, true);
- admin2.namespaces().deleteNamespace(nonReplicatedNamespace, true);
+ Awaitility.await().atMost(Duration.ofSeconds(30)).ignoreExceptions().untilAsserted(() -> {
+ admin2.namespaces().deleteNamespace(replicatedNamespace, true);
+ });
+ Awaitility.await().atMost(Duration.ofSeconds(30)).ignoreExceptions().untilAsserted(() -> {
+ admin2.namespaces().deleteNamespace(nonReplicatedNamespace, true);
+ });
+ Awaitility.await().atMost(Duration.ofSeconds(30)).ignoreExceptions().untilAsserted(() -> {
+ admin2.namespaces().deleteNamespace(sourceClusterAlwaysSchemaCompatibleNamespace, true);
+ });
}
}
@Override
protected void cleanup() throws Exception {
// cleanup pulsar resources.
- cleanupPulsarResources();
+ // Wrap in try-catch to ensure brokers, ZK, and BK are always shut down even if
+ // namespace deletion fails (e.g., topics in transitional state during async replication cleanup).
+ try {
+ cleanupPulsarResources();
+ } catch (Exception e) {
+ log.warn("Failed to cleanup Pulsar resources during shutdown, "
+ + "continuing with broker/ZK/BK shutdown", e);
+ }
// shutdown.
markCurrentSetupNumberCleaned();