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/03/10 09:19:58 UTC
(pulsar) branch branch-4.1 updated (d11bb98f0c7 -> a2445644a8e)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from d11bb98f0c7 [fix][sec] Upgrade Jackson version to 2.18.6 (#25264)
new 418dddc924a [fix][broker] Support namespace unsubscribe when bundles are unloaded (#25276)
new 03d5f81b316 [fix][sec] Bump org.apache.zookeeper:zookeeper from 3.9.4 to 3.9.5 (#25303)
new a2445644a8e [fix][offload] Close all resources in BlobStoreBackedReadHandleImplV2.closeAsync (#25296)
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
distribution/server/src/assemble/LICENSE.bin.txt | 6 +-
pom.xml | 2 +-
.../pulsar/broker/admin/impl/NamespacesBase.java | 156 +++++++--------
.../apache/pulsar/broker/admin/v1/Namespaces.java | 29 ++-
.../apache/pulsar/broker/admin/v2/Namespaces.java | 29 ++-
.../apache/pulsar/broker/admin/AdminApiTest.java | 215 +++++++++++++++++++++
.../impl/BlobStoreBackedReadHandleImplV2.java | 30 ++-
7 files changed, 358 insertions(+), 109 deletions(-)
(pulsar) 01/03: [fix][broker] Support namespace unsubscribe when bundles are unloaded (#25276)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 418dddc924a1745fe8fae130a8c1c1ab395b2160
Author: Hao Zhang <zh...@cmss.chinamobile.com>
AuthorDate: Fri Mar 6 02:27:14 2026 +0800
[fix][broker] Support namespace unsubscribe when bundles are unloaded (#25276)
(cherry picked from commit f83251a57fd5f4bc20cd47ec7b950d6321d03488)
---
.../pulsar/broker/admin/impl/NamespacesBase.java | 156 +++++++--------
.../apache/pulsar/broker/admin/v1/Namespaces.java | 29 ++-
.../apache/pulsar/broker/admin/v2/Namespaces.java | 29 ++-
.../apache/pulsar/broker/admin/AdminApiTest.java | 215 +++++++++++++++++++++
4 files changed, 328 insertions(+), 101 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
index 0cb4f9a493d..f28b07d8a9a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
@@ -1698,70 +1698,51 @@ public abstract class NamespacesBase extends AdminResource {
subscription, namespaceName, bundleRange);
}
- protected void internalUnsubscribeNamespace(AsyncResponse asyncResponse, String subscription,
- boolean authoritative) {
- validateNamespaceOperation(namespaceName, NamespaceOperation.UNSUBSCRIBE);
+ protected CompletableFuture<Void> internalUnsubscribeNamespaceAsync(String subscription,
+ boolean authoritative) {
checkNotNull(subscription, "Subscription should not be null");
- final List<CompletableFuture<Void>> futures = new ArrayList<>();
- try {
- NamespaceBundles bundles = pulsar().getNamespaceService().getNamespaceBundleFactory()
- .getBundles(namespaceName);
- for (NamespaceBundle nsBundle : bundles.getBundles()) {
- // check if the bundle is owned by any broker, if not then there are no subscriptions
- if (pulsar().getNamespaceService().checkOwnershipPresent(nsBundle)) {
- futures.add(pulsar().getAdminClient().namespaces().unsubscribeNamespaceBundleAsync(
- namespaceName.toString(), nsBundle.getBundleRange(), subscription));
- }
- }
- } catch (WebApplicationException wae) {
- asyncResponse.resume(wae);
- return;
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- return;
- }
-
- FutureUtil.waitForAll(futures).handle((result, exception) -> {
- if (exception != null) {
- log.warn("[{}] Failed to unsubscribe {} on the bundles for namespace {}: {}", clientAppId(),
- subscription, namespaceName, exception.getCause().getMessage());
- if (exception.getCause() instanceof PulsarAdminException) {
- asyncResponse.resume(new RestException((PulsarAdminException) exception.getCause()));
- return null;
- } else {
- asyncResponse.resume(new RestException(exception.getCause()));
- return null;
- }
- }
- log.info("[{}] Successfully unsubscribed {} on all the bundles for namespace {}", clientAppId(),
- subscription, namespaceName);
- asyncResponse.resume(Response.noContent().build());
- return null;
- });
+ return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.UNSUBSCRIBE)
+ .thenCompose(__ -> pulsar().getNamespaceService().getNamespaceBundleFactory()
+ .getBundlesAsync(namespaceName))
+ .thenCompose(bundles -> {
+ final List<CompletableFuture<Void>> futures = new ArrayList<>();
+ for (NamespaceBundle nsBundle : bundles.getBundles()) {
+ try {
+ futures.add(pulsar().getAdminClient().namespaces().unsubscribeNamespaceBundleAsync(
+ namespaceName.toString(), nsBundle.getBundleRange(), subscription));
+ } catch (PulsarServerException e) {
+ return CompletableFuture.failedFuture(e);
+ }
+ }
+ return FutureUtil.waitForAll(futures);
+ }).thenRun(() -> log.info("[{}] Successfully unsubscribed {} on all the bundles for namespace {}",
+ clientAppId(), subscription, namespaceName));
}
@SuppressWarnings("deprecation")
- protected void internalUnsubscribeNamespaceBundle(String subscription, String bundleRange, boolean authoritative) {
- validateNamespaceOperation(namespaceName, NamespaceOperation.UNSUBSCRIBE);
+ protected CompletableFuture<Void> internalUnsubscribeNamespaceBundleAsync(String subscription, String bundleRange,
+ boolean authoritative) {
checkNotNull(subscription, "Subscription should not be null");
checkNotNull(bundleRange, "BundleRange should not be null");
- Policies policies = getNamespacePolicies(namespaceName);
-
- if (namespaceName.isGlobal()) {
- // check cluster ownership for a given global namespace: redirect if peer-cluster owns it
- validateGlobalNamespaceOwnership(namespaceName);
- } else {
- validateClusterOwnership(namespaceName.getCluster());
- validateClusterForTenant(namespaceName.getTenant(), namespaceName.getCluster());
- }
-
- validateNamespaceBundleOwnership(namespaceName, policies.bundles, bundleRange, authoritative, true);
-
- unsubscribe(namespaceName, bundleRange, subscription);
- log.info("[{}] Successfully unsubscribed {} on namespace bundle {}/{}", clientAppId(), subscription,
- namespaceName, bundleRange);
+ return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.UNSUBSCRIBE)
+ .thenCompose(__ -> {
+ if (namespaceName.isGlobal()) {
+ // check cluster ownership for a given global namespace: redirect if peer-cluster owns it
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ }
+ return validateClusterOwnershipAsync(namespaceName.getCluster())
+ .thenCompose(unused -> validateClusterForTenantAsync(namespaceName.getTenant(),
+ namespaceName.getCluster()));
+ })
+ .thenCompose(__ -> getNamespacePoliciesAsync(namespaceName))
+ .thenCompose(policies ->
+ validateNamespaceBundleOwnershipAsync(namespaceName, policies.bundles, bundleRange,
+ authoritative, false))
+ .thenCompose(bundle -> unsubscribeAsync(bundle, subscription))
+ .thenRun(() -> log.info("[{}] Successfully unsubscribed {} on namespace bundle {}/{}",
+ clientAppId(), subscription, namespaceName, bundleRange));
}
protected void internalSetSubscriptionAuthMode(SubscriptionAuthMode subscriptionAuthMode) {
@@ -1984,32 +1965,45 @@ public abstract class NamespacesBase extends AdminResource {
}
}
- private void unsubscribe(NamespaceName nsName, String bundleRange, String subscription) {
- try {
- List<Topic> topicList = pulsar().getBrokerService().getAllTopicsFromNamespaceBundle(nsName.toString(),
- nsName.toString() + "/" + bundleRange);
- List<CompletableFuture<Void>> futures = new ArrayList<>();
- if (subscription.startsWith(pulsar().getConfiguration().getReplicatorPrefix())) {
- throw new RestException(Status.PRECONDITION_FAILED, "Cannot unsubscribe a replication cursor");
- } else {
- for (Topic topic : topicList) {
- Subscription sub = topic.getSubscription(subscription);
- if (sub != null) {
- futures.add(sub.delete());
- }
- }
- }
- FutureUtil.waitForAll(futures).get();
- } catch (RestException re) {
- throw re;
- } catch (Exception e) {
- log.error("[{}] Failed to unsubscribe {} for namespace {}/{}", clientAppId(), subscription,
- nsName.toString(), bundleRange, e);
- if (e.getCause() instanceof SubscriptionBusyException) {
- throw new RestException(Status.PRECONDITION_FAILED, "Subscription has active connected consumers");
- }
- throw new RestException(e.getCause());
+ private CompletableFuture<Void> unsubscribeAsync(NamespaceBundle bundle, String subscription) {
+ if (subscription.startsWith(pulsar().getConfiguration().getReplicatorPrefix())) {
+ return CompletableFuture.failedFuture(
+ new RestException(Status.PRECONDITION_FAILED, "Cannot unsubscribe a replication cursor"));
}
+
+ return pulsar().getNamespaceService().getOwnedTopicListForNamespaceBundle(bundle)
+ .thenCompose(topicsInBundle -> {
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ for (String topic : topicsInBundle) {
+ TopicName topicName = TopicName.get(topic);
+ if (pulsar().getBrokerService().isSystemTopic(topicName)) {
+ continue;
+ }
+ futures.add(pulsar().getBrokerService().getTopic(topicName.toString(), false)
+ .thenCompose(optTopic -> {
+ if (optTopic.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ Topic loaded = optTopic.get();
+ Subscription sub = loaded.getSubscription(subscription);
+ if (sub == null) {
+ return CompletableFuture.completedFuture(null);
+ }
+ return sub.delete();
+ }));
+ }
+ return FutureUtil.waitForAll(futures);
+ }).exceptionally(ex -> {
+ Throwable cause = FutureUtil.unwrapCompletionException(ex);
+ if (cause instanceof RestException) {
+ throw (RestException) cause;
+ }
+ if (cause instanceof SubscriptionBusyException) {
+ throw new RestException(Status.PRECONDITION_FAILED,
+ "Subscription has active connected consumers");
+ }
+ throw new RestException(cause);
+ });
}
protected BundlesData validateBundlesData(BundlesData initialBundles) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java
index af714ac1aee..276ea457d2c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java
@@ -1408,14 +1408,15 @@ public class Namespaces extends NamespacesBase {
@PathParam("property") String property, @PathParam("cluster") String cluster,
@PathParam("namespace") String namespace, @PathParam("subscription") String subscription,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
- try {
- validateNamespaceName(property, cluster, namespace);
- internalUnsubscribeNamespace(asyncResponse, subscription, authoritative);
- } catch (WebApplicationException wae) {
- asyncResponse.resume(wae);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ validateNamespaceName(property, cluster, namespace);
+ internalUnsubscribeNamespaceAsync(subscription, authoritative)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to unsubscribe {} on namespace {}", clientAppId(),
+ subscription, namespaceName, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@POST
@@ -1424,12 +1425,20 @@ public class Namespaces extends NamespacesBase {
@ApiResponses(value = {
@ApiResponse(code = 403, message = "Don't have admin or operate permission on the namespace"),
@ApiResponse(code = 404, message = "Namespace does not exist") })
- public void unsubscribeNamespaceBundle(@PathParam("property") String property, @PathParam("cluster") String cluster,
+ public void unsubscribeNamespaceBundle(@Suspended final AsyncResponse asyncResponse,
+ @PathParam("property") String property, @PathParam("cluster") String cluster,
@PathParam("namespace") String namespace, @PathParam("subscription") String subscription,
@PathParam("bundle") String bundleRange,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateNamespaceName(property, cluster, namespace);
- internalUnsubscribeNamespaceBundle(subscription, bundleRange, authoritative);
+ internalUnsubscribeNamespaceBundleAsync(subscription, bundleRange, authoritative)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to unsubscribe {} on namespace bundle {}/{}", clientAppId(),
+ subscription, namespaceName, bundleRange, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@POST
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
index a8eea7cfb5b..5058eccc401 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
@@ -1585,14 +1585,15 @@ public class Namespaces extends NamespacesBase {
@PathParam("namespace") String namespace,
@PathParam("subscription") String subscription,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
- try {
- validateNamespaceName(tenant, namespace);
- internalUnsubscribeNamespace(asyncResponse, subscription, authoritative);
- } catch (WebApplicationException wae) {
- asyncResponse.resume(wae);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ validateNamespaceName(tenant, namespace);
+ internalUnsubscribeNamespaceAsync(subscription, authoritative)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to unsubscribe {} on namespace {}", clientAppId(),
+ subscription, namespaceName, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@POST
@@ -1602,12 +1603,20 @@ public class Namespaces extends NamespacesBase {
@ApiResponse(code = 204, message = "Operation successful"),
@ApiResponse(code = 403, message = "Don't have admin or operate permission on the namespace"),
@ApiResponse(code = 404, message = "Namespace does not exist") })
- public void unsubscribeNamespaceBundle(@PathParam("tenant") String tenant,
+ public void unsubscribeNamespaceBundle(@Suspended final AsyncResponse asyncResponse,
+ @PathParam("tenant") String tenant,
@PathParam("namespace") String namespace, @PathParam("subscription") String subscription,
@PathParam("bundle") String bundleRange,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateNamespaceName(tenant, namespace);
- internalUnsubscribeNamespaceBundle(subscription, bundleRange, authoritative);
+ internalUnsubscribeNamespaceBundleAsync(subscription, bundleRange, authoritative)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to unsubscribe {} on namespace bundle {}/{}", clientAppId(),
+ subscription, namespaceName, bundleRange, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@POST
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
index 663a6f0a419..8dabc75e52d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
@@ -2339,6 +2339,221 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
new ArrayList<>());
}
+ @Test(dataProvider = "numBundles")
+ public void testUnsubscribeNamespaceBundleOnUnloadedBundle(Integer numBundles) throws Exception {
+ String namespace = "prop-xyz/ns-unsub-bundle";
+ admin.namespaces().createNamespace(namespace, numBundles);
+
+ String topic = "persistent://" + namespace + "/t1";
+ String subscription = "sub1";
+ String otherSubscription = "sub2";
+
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionName(subscription)
+ .subscribe();
+
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionName(otherSubscription)
+ .subscribe();
+
+ consumer1.close();
+ consumer2.close();
+
+ NamespaceBundle bundle =
+ pulsar.getNamespaceService().getNamespaceBundleFactory().getBundle(TopicName.get(topic));
+ admin.namespaces().unloadNamespaceBundle(namespace, bundle.getBundleRange());
+
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .pollInterval(200, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> {
+ assertFalse(pulsar.getNamespaceService().isServiceUnitOwned(bundle),
+ "Bundle should not be owned by current broker");
+ assertFalse(otherPulsar.getNamespaceService().isServiceUnitOwned(bundle),
+ "Bundle should not be owned by other broker");
+ });
+
+ admin.namespaces().unsubscribeNamespaceBundle(namespace, bundle.getBundleRange(), subscription);
+
+ List<String> subscriptions =
+ admin.topics().getSubscriptions(topic).stream().sorted().toList();
+ assertEquals(subscriptions, List.of(otherSubscription));
+ }
+
+ @Test(dataProvider = "numBundles")
+ public void testUnsubscribeNamespaceOnUnloadedBundle(Integer numBundles) throws Exception {
+ String namespace = "prop-xyz/ns-unsub-namespace";
+ admin.namespaces().createNamespace(namespace, numBundles);
+
+ String topic = "persistent://" + namespace + "/t1";
+ String subscription = "sub1";
+ String otherSubscription = "sub2";
+
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionName(subscription)
+ .subscribe();
+
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionName(otherSubscription)
+ .subscribe();
+
+ consumer1.close();
+ consumer2.close();
+
+ NamespaceBundle bundle =
+ pulsar.getNamespaceService().getNamespaceBundleFactory().getBundle(TopicName.get(topic));
+ admin.namespaces().unloadNamespaceBundle(namespace, bundle.getBundleRange());
+
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .pollInterval(200, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> {
+ assertFalse(pulsar.getNamespaceService().isServiceUnitOwned(bundle),
+ "Bundle should not be owned by current broker");
+ assertFalse(otherPulsar.getNamespaceService().isServiceUnitOwned(bundle),
+ "Bundle should not be owned by other broker");
+ });
+
+ admin.namespaces().unsubscribeNamespace(namespace, subscription);
+
+ List<String> subscriptions =
+ admin.topics().getSubscriptions(topic).stream().sorted().toList();
+ assertEquals(subscriptions, List.of(otherSubscription));
+ }
+
+ @Test(dataProvider = "numBundles")
+ public void testUnsubscribeNamespaceOnUnloadedBundleWithPartitionedTopic(Integer numBundles) throws Exception {
+ String namespace = "prop-xyz/ns-unsub-partitioned";
+ admin.namespaces().createNamespace(namespace, numBundles);
+
+ String topic = "persistent://" + namespace + "/pt";
+ int partitions = 3;
+ admin.topics().createPartitionedTopic(topic, partitions);
+
+ String subscription = "sub1";
+ String otherSubscription = "sub2";
+
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionName(subscription)
+ .subscribe();
+
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionName(otherSubscription)
+ .subscribe();
+
+ consumer1.close();
+ consumer2.close();
+
+ Set<NamespaceBundle> bundles = new HashSet<>();
+ for (int i = 0; i < partitions; i++) {
+ String partitionTopic = TopicName.get(topic).getPartition(i).toString();
+ bundles.add(pulsar.getNamespaceService().getNamespaceBundleFactory()
+ .getBundle(TopicName.get(partitionTopic)));
+ }
+ for (NamespaceBundle bundle : bundles) {
+ admin.namespaces().unloadNamespaceBundle(namespace, bundle.getBundleRange());
+ }
+
+ for (NamespaceBundle bundle : bundles) {
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .pollInterval(200, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> {
+ assertFalse(pulsar.getNamespaceService().isServiceUnitOwned(bundle),
+ "Bundle should not be owned by current broker");
+ assertFalse(otherPulsar.getNamespaceService().isServiceUnitOwned(bundle),
+ "Bundle should not be owned by other broker");
+ });
+ }
+
+ admin.namespaces().unsubscribeNamespace(namespace, subscription);
+
+ for (int i = 0; i < partitions; i++) {
+ String partitionTopic = TopicName.get(topic).getPartition(i).toString();
+ List<String> subs = admin.topics().getSubscriptions(partitionTopic).stream().sorted().toList();
+ assertEquals(subs, List.of(otherSubscription));
+ }
+ }
+
+ @Test
+ public void testUnsubscribeNamespaceOnUnloadedBundleWithMultiTopicCrossBundle() throws Exception {
+ String namespace = "prop-xyz/ns-unsub-cross-bundle";
+ admin.namespaces().createNamespace(namespace, 4);
+
+ String subscription = "sub1";
+ String otherSubscription = "sub2";
+
+ String topic1 = null;
+ String topic2 = null;
+ NamespaceBundle bundle1 = null;
+ NamespaceBundle bundle2 = null;
+ for (int i = 0; i < 50; i++) {
+ String candidate = "persistent://" + namespace + "/t-" + i;
+ @Cleanup
+ Consumer<byte[]> c = pulsarClient.newConsumer()
+ .topic(candidate)
+ .subscriptionName(subscription)
+ .subscribe();
+ c.close();
+ NamespaceBundle bundle =
+ pulsar.getNamespaceService().getNamespaceBundleFactory().getBundle(TopicName.get(candidate));
+ if (topic1 == null) {
+ topic1 = candidate;
+ bundle1 = bundle;
+ continue;
+ }
+ if (!bundle.equals(bundle1)) {
+ topic2 = candidate;
+ bundle2 = bundle;
+ break;
+ }
+ }
+ assertNotNull(topic1);
+ assertNotNull(topic2);
+ assertNotNull(bundle1);
+ assertNotNull(bundle2);
+
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topic(topic1)
+ .subscriptionName(otherSubscription)
+ .subscribe();
+
+ Consumer<byte[]> consumer2 = pulsarClient.newConsumer()
+ .topic(topic2)
+ .subscriptionName(otherSubscription)
+ .subscribe();
+ consumer1.close();
+ consumer2.close();
+
+ admin.namespaces().unloadNamespaceBundle(namespace, bundle1.getBundleRange());
+ admin.namespaces().unloadNamespaceBundle(namespace, bundle2.getBundleRange());
+
+ for (NamespaceBundle bundle : List.of(bundle1, bundle2)) {
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .pollInterval(200, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> {
+ assertFalse(pulsar.getNamespaceService().isServiceUnitOwned(bundle),
+ "Bundle should not be owned by current broker");
+ assertFalse(otherPulsar.getNamespaceService().isServiceUnitOwned(bundle),
+ "Bundle should not be owned by other broker");
+ });
+ }
+
+ admin.namespaces().unsubscribeNamespace(namespace, subscription);
+
+ List<String> subs1 = admin.topics().getSubscriptions(topic1).stream().sorted().toList();
+ List<String> subs2 = admin.topics().getSubscriptions(topic2).stream().sorted().toList();
+ assertEquals(subs1, List.of(otherSubscription));
+ assertEquals(subs2, List.of(otherSubscription));
+ }
+
private List<MessageId> publishMessagesOnPersistentTopic(String topicName, int messages) throws Exception {
return publishMessagesOnPersistentTopic(topicName, messages, 0, false);
}
(pulsar) 03/03: [fix][offload] Close all resources in BlobStoreBackedReadHandleImplV2.closeAsync (#25296)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a2445644a8ecc58cfc41d3570b408a1659909809
Author: Sanjay Malakar <ia...@gmail.com>
AuthorDate: Mon Mar 9 19:37:33 2026 -0700
[fix][offload] Close all resources in BlobStoreBackedReadHandleImplV2.closeAsync (#25296)
(cherry picked from commit c83cafc12eb1db4c3f17c4f0e47ebb3b8403fe56)
---
.../impl/BlobStoreBackedReadHandleImplV2.java | 30 +++++++++++++++++++---
1 file changed, 26 insertions(+), 4 deletions(-)
diff --git a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImplV2.java b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImplV2.java
index 3fffd18a5d8..1ac5047ba71 100644
--- a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImplV2.java
+++ b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImplV2.java
@@ -133,15 +133,37 @@ public class BlobStoreBackedReadHandleImplV2 implements ReadHandle {
CompletableFuture<Void> promise = closeFuture.get();
executor.execute(() -> {
try {
+ IOException first = null;
for (OffloadIndexBlockV2 indexBlock : indices) {
- indexBlock.close();
+ try {
+ indexBlock.close();
+ } catch (IOException e) {
+ if (first == null) {
+ first = e;
+ } else {
+ first.addSuppressed(e);
+ }
+ }
}
for (DataInputStream dataStream : dataStreams) {
- dataStream.close();
+ try {
+ dataStream.close();
+ } catch (IOException e) {
+ if (first == null) {
+ first = e;
+ } else {
+ first.addSuppressed(e);
+ }
+ }
+ }
+ state = State.Closed;
+ if (first != null) {
+ promise.completeExceptionally(first);
+ } else {
+ promise.complete(null);
}
+ } catch (Throwable t) {
state = State.Closed;
- promise.complete(null);
- } catch (IOException t) {
promise.completeExceptionally(t);
}
});
(pulsar) 02/03: [fix][sec] Bump org.apache.zookeeper:zookeeper from 3.9.4 to 3.9.5 (#25303)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 03d5f81b31677282789a5fd55c783c7bea61f544
Author: dependabot[bot] <49...@users.noreply.github.com>
AuthorDate: Tue Mar 10 10:34:58 2026 +0800
[fix][sec] Bump org.apache.zookeeper:zookeeper from 3.9.4 to 3.9.5 (#25303)
(cherry picked from commit 5081c02e7f58288ff6842cc66032acb0a6d7577b)
---
distribution/server/src/assemble/LICENSE.bin.txt | 6 +++---
pom.xml | 2 +-
2 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt
index 4879d43f60d..e8bcb233b70 100644
--- a/distribution/server/src/assemble/LICENSE.bin.txt
+++ b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -501,9 +501,9 @@ The Apache Software License, Version 2.0
- io.vertx-vertx-web-4.5.24.jar
- io.vertx-vertx-web-common-4.5.24.jar
* Apache ZooKeeper
- - org.apache.zookeeper-zookeeper-3.9.4.jar
- - org.apache.zookeeper-zookeeper-jute-3.9.4.jar
- - org.apache.zookeeper-zookeeper-prometheus-metrics-3.9.4.jar
+ - org.apache.zookeeper-zookeeper-3.9.5.jar
+ - org.apache.zookeeper-zookeeper-jute-3.9.5.jar
+ - org.apache.zookeeper-zookeeper-prometheus-metrics-3.9.5.jar
* Snappy Java
- org.xerial.snappy-snappy-java-1.1.10.8.jar
* Google HTTP Client
diff --git a/pom.xml b/pom.xml
index c314156fc4c..7c763aa0874 100644
--- a/pom.xml
+++ b/pom.xml
@@ -183,7 +183,7 @@ flexible messaging model and an intuitive client API.</description>
<commons-compress.version>1.28.0</commons-compress.version>
<bookkeeper.version>4.17.3</bookkeeper.version>
- <zookeeper.version>3.9.4</zookeeper.version>
+ <zookeeper.version>3.9.5</zookeeper.version>
<commons-cli.version>1.11.0</commons-cli.version>
<commons-text.version>1.14.0</commons-text.version>
<snappy.version>1.1.10.8</snappy.version> <!-- ZooKeeper server -->