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/15 16:19:26 UTC
(pulsar) branch branch-4.0 updated (fb5f211ee9b -> c5a05998424)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from fb5f211ee9b [improve][build][branch-4.0] Support docker.golang.image/GOLANG_IMAGE in latest-version-image
new f0e98a20bbc [fix][build][branch-4.0] Fix broken compilation after cherry-picking #25400
new eefa9f0fc04 Revert "[fix][admin] Refactor namespace migration operation to async in rest api (#25478)"
new bc8449deb3a Revert "[fix][broker] Fix backlog clearing for unloaded namespace bundles (#25272)"
new 125d1d194f4 [fix][admin] Refactor namespace anti affinity group sync operations to async in rest api (#25086)
new 9a969707abd [fix][client] Fix async APIs to return failed futures on validation errors (#25287)
new 0a717fb5765 [fix][broker] Return failed future instead of throwing exception in async methods (#25289)
new b960c28983b [fix][admin] Refactor namespace migration operation to async in rest api (#25478)
new eb79b2d0130 [fix][broker] Fix backlog clearing for unloaded namespace bundles (#25272)
new bad69d87095 [fix][build][branch-4.0] Fix missing exclusion in cherry-picking #25264
new 98119ce9b21 [fix][build] Fix license file for shell distribution
new c5a05998424 [cleanup][build] Bumped version to 4.0.10-SNAPSHOT
The 11 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:
bouncy-castle/bc/pom.xml | 2 +-
bouncy-castle/bcfips-include-test/pom.xml | 2 +-
bouncy-castle/bcfips/pom.xml | 2 +-
bouncy-castle/pom.xml | 2 +-
buildtools/pom.xml | 4 +-
distribution/io/pom.xml | 2 +-
distribution/offloaders/pom.xml | 2 +-
distribution/pom.xml | 2 +-
distribution/server/pom.xml | 2 +-
distribution/shell/pom.xml | 2 +-
distribution/shell/src/assemble/LICENSE.bin.txt | 2 +-
docker/pom.xml | 2 +-
docker/pulsar-all/pom.xml | 2 +-
docker/pulsar/pom.xml | 2 +-
jclouds-shaded/pom.xml | 2 +-
jetcd-core-shaded/pom.xml | 2 +-
managed-ledger/pom.xml | 2 +-
microbench/pom.xml | 2 +-
pom.xml | 8 +-
pulsar-bom/pom.xml | 4 +-
pulsar-broker-auth-athenz/pom.xml | 2 +-
pulsar-broker-auth-oidc/pom.xml | 2 +-
pulsar-broker-auth-sasl/pom.xml | 2 +-
pulsar-broker-common/pom.xml | 2 +-
.../authorization/PulsarAuthorizationProvider.java | 2 +-
.../broker/resources/LocalPoliciesResources.java | 6 +-
pulsar-broker/pom.xml | 2 +-
.../apache/pulsar/broker/admin/AdminResource.java | 10 +-
.../pulsar/broker/admin/impl/BrokersBase.java | 4 +-
.../pulsar/broker/admin/impl/NamespacesBase.java | 198 ++++++++-------------
.../broker/admin/impl/PersistentTopicsBase.java | 37 ++--
.../apache/pulsar/broker/admin/v1/Namespaces.java | 112 ++++++++++--
.../apache/pulsar/broker/admin/v2/Namespaces.java | 58 ++++--
.../pulsar/broker/delayed/bucket/Bucket.java | 5 +-
.../channel/ServiceUnitStateChannelImpl.java | 2 +-
.../pulsar/broker/namespace/NamespaceService.java | 16 +-
.../pulsar/broker/service/BrokerService.java | 8 +-
.../SystemTopicBasedTopicPoliciesService.java | 4 +-
.../org/apache/pulsar/broker/service/Topic.java | 4 +-
.../service/nonpersistent/NonPersistentTopic.java | 3 +-
.../prometheus/PrometheusMetricsGenerator.java | 3 +-
.../pulsar/broker/systopic/SystemTopicClient.java | 3 +-
.../pulsar/broker/web/PulsarWebResource.java | 6 +-
.../compaction/PulsarCompactionServiceFactory.java | 10 +-
.../compaction/StrategicTwoPhaseCompactor.java | 2 +-
.../apache/pulsar/broker/admin/NamespacesTest.java | 80 +++++++++
.../pulsar/broker/admin/NamespacesV2Test.java | 20 ++-
.../pulsar/client/impl/MessageChunkingTest.java | 37 ----
pulsar-cli-utils/pom.xml | 2 +-
pulsar-client-admin-api/pom.xml | 2 +-
pulsar-client-admin-shaded/pom.xml | 2 +-
pulsar-client-admin/pom.xml | 2 +-
pulsar-client-all/pom.xml | 2 +-
pulsar-client-api/pom.xml | 2 +-
pulsar-client-auth-athenz/pom.xml | 2 +-
pulsar-client-auth-sasl/pom.xml | 2 +-
pulsar-client-dependencies-minimized/pom.xml | 2 +-
pulsar-client-messagecrypto-bc/pom.xml | 2 +-
pulsar-client-shaded/pom.xml | 2 +-
pulsar-client-tools-api/pom.xml | 2 +-
pulsar-client-tools-customcommand-example/pom.xml | 2 +-
pulsar-client-tools-test/pom.xml | 2 +-
pulsar-client-tools/pom.xml | 2 +-
pulsar-client/pom.xml | 2 +-
.../apache/pulsar/client/impl/ConsumerBase.java | 11 +-
.../client/impl/MultiTopicsConsumerImpl.java | 4 +-
.../pulsar/client/impl/ProducerBuilderImpl.java | 6 +-
.../client/impl/PulsarChannelInitializer.java | 12 +-
.../client/impl/MultiTopicsConsumerImplTest.java | 11 ++
.../client/impl/ProducerBuilderImplTest.java | 20 +++
pulsar-common/pom.xml | 2 +-
.../org/apache/pulsar/common/util/FutureUtil.java | 21 ++-
.../apache/pulsar/common/util/SecurityUtility.java | 3 +-
.../pulsar/common/util/netty/ChannelFutures.java | 6 +-
.../pulsar/common/util/netty/NettyFutureUtil.java | 10 +-
.../common/util/netty/ChannelFuturesTest.java | 12 +-
pulsar-config-validation/pom.xml | 2 +-
pulsar-docs-tools/pom.xml | 2 +-
pulsar-functions/api-java/pom.xml | 2 +-
pulsar-functions/instance/pom.xml | 2 +-
pulsar-functions/java-examples-builtin/pom.xml | 2 +-
pulsar-functions/java-examples/pom.xml | 2 +-
pulsar-functions/localrun-shaded/pom.xml | 2 +-
pulsar-functions/localrun/pom.xml | 2 +-
pulsar-functions/pom.xml | 2 +-
pulsar-functions/proto/pom.xml | 2 +-
pulsar-functions/runtime-all/pom.xml | 2 +-
pulsar-functions/runtime/pom.xml | 2 +-
pulsar-functions/secrets/pom.xml | 2 +-
pulsar-functions/utils/pom.xml | 2 +-
pulsar-functions/worker/pom.xml | 2 +-
pulsar-io/aerospike/pom.xml | 2 +-
pulsar-io/alluxio/pom.xml | 2 +-
pulsar-io/aws/pom.xml | 2 +-
pulsar-io/azure-data-explorer/pom.xml | 2 +-
pulsar-io/batch-data-generator/pom.xml | 2 +-
pulsar-io/batch-discovery-triggerers/pom.xml | 2 +-
pulsar-io/canal/pom.xml | 2 +-
pulsar-io/cassandra/pom.xml | 2 +-
pulsar-io/common/pom.xml | 2 +-
pulsar-io/core/pom.xml | 2 +-
pulsar-io/data-generator/pom.xml | 2 +-
pulsar-io/debezium/core/pom.xml | 2 +-
pulsar-io/debezium/mongodb/pom.xml | 2 +-
pulsar-io/debezium/mssql/pom.xml | 2 +-
pulsar-io/debezium/mysql/pom.xml | 2 +-
pulsar-io/debezium/oracle/pom.xml | 2 +-
pulsar-io/debezium/pom.xml | 2 +-
pulsar-io/debezium/postgres/pom.xml | 2 +-
pulsar-io/docs/pom.xml | 2 +-
pulsar-io/dynamodb/pom.xml | 2 +-
pulsar-io/elastic-search/pom.xml | 2 +-
pulsar-io/file/pom.xml | 2 +-
pulsar-io/flume/pom.xml | 2 +-
pulsar-io/hbase/pom.xml | 2 +-
pulsar-io/hdfs3/pom.xml | 2 +-
pulsar-io/http/pom.xml | 2 +-
pulsar-io/influxdb/pom.xml | 2 +-
pulsar-io/jdbc/clickhouse/pom.xml | 2 +-
pulsar-io/jdbc/core/pom.xml | 2 +-
pulsar-io/jdbc/mariadb/pom.xml | 2 +-
pulsar-io/jdbc/openmldb/pom.xml | 2 +-
pulsar-io/jdbc/pom.xml | 2 +-
pulsar-io/jdbc/postgres/pom.xml | 2 +-
pulsar-io/jdbc/sqlite/pom.xml | 2 +-
pulsar-io/kafka-connect-adaptor-nar/pom.xml | 2 +-
pulsar-io/kafka-connect-adaptor/pom.xml | 2 +-
pulsar-io/kafka/pom.xml | 2 +-
pulsar-io/kinesis-kpl-shaded/pom.xml | 2 +-
pulsar-io/kinesis/pom.xml | 2 +-
pulsar-io/mongo/pom.xml | 2 +-
pulsar-io/netty/pom.xml | 2 +-
pulsar-io/nsq/pom.xml | 2 +-
pulsar-io/pom.xml | 2 +-
pulsar-io/rabbitmq/pom.xml | 2 +-
pulsar-io/redis/pom.xml | 2 +-
pulsar-io/solr/pom.xml | 2 +-
pulsar-io/twitter/pom.xml | 2 +-
pulsar-metadata/pom.xml | 2 +-
pulsar-opentelemetry/pom.xml | 2 +-
.../bookkeeper-storage/pom.xml | 2 +-
pulsar-package-management/core/pom.xml | 2 +-
.../filesystem-storage/pom.xml | 2 +-
pulsar-package-management/pom.xml | 2 +-
pulsar-proxy/pom.xml | 2 +-
pulsar-testclient/pom.xml | 2 +-
pulsar-transaction/common/pom.xml | 2 +-
pulsar-transaction/coordinator/pom.xml | 2 +-
pulsar-transaction/pom.xml | 2 +-
pulsar-websocket/pom.xml | 2 +-
structured-event-log/pom.xml | 2 +-
testmocks/pom.xml | 2 +-
tests/bc_2_0_0/pom.xml | 2 +-
tests/bc_2_0_1/pom.xml | 2 +-
tests/bc_2_6_0/pom.xml | 2 +-
tests/docker-images/java-test-functions/pom.xml | 2 +-
tests/docker-images/java-test-image/pom.xml | 2 +-
tests/docker-images/java-test-plugins/pom.xml | 2 +-
tests/docker-images/latest-version-image/pom.xml | 2 +-
tests/docker-images/pom.xml | 2 +-
tests/integration/pom.xml | 2 +-
tests/pom.xml | 2 +-
tests/pulsar-client-admin-shade-test/pom.xml | 2 +-
tests/pulsar-client-all-shade-test/pom.xml | 2 +-
tests/pulsar-client-shade-test/pom.xml | 2 +-
tiered-storage/file-system/pom.xml | 2 +-
tiered-storage/jcloud/pom.xml | 2 +-
tiered-storage/pom.xml | 2 +-
168 files changed, 614 insertions(+), 410 deletions(-)
(pulsar) 10/11: [fix][build] Fix license file for shell distribution
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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 98119ce9b214e43bd9e1eacdbe815257edb18ddf
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Wed Apr 15 19:13:15 2026 +0300
[fix][build] Fix license file for shell distribution
- fix for cherry-picking of #25400
---
distribution/shell/src/assemble/LICENSE.bin.txt | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/distribution/shell/src/assemble/LICENSE.bin.txt b/distribution/shell/src/assemble/LICENSE.bin.txt
index 509e99ccf8c..4e2820a57fd 100644
--- a/distribution/shell/src/assemble/LICENSE.bin.txt
+++ b/distribution/shell/src/assemble/LICENSE.bin.txt
@@ -324,7 +324,7 @@ The Apache Software License, Version 2.0
- jackson-datatype-jdk8-2.18.6.jar
- jackson-datatype-jsr310-2.18.6.jar
- jackson-module-parameter-names-2.18.6.jar
- * Caffeine -- caffeine-3.2.3.jar
+ * Caffeine -- caffeine-2.9.1.jar
* Conscrypt -- conscrypt-openjdk-uber-2.5.2.jar
* Gson
- gson-2.13.2.jar
(pulsar) 11/11: [cleanup][build] Bumped version to 4.0.10-SNAPSHOT
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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c5a0599842464332239461a14742066ab88331ee
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Wed Apr 15 19:16:38 2026 +0300
[cleanup][build] Bumped version to 4.0.10-SNAPSHOT
Signed-off-by: Lari Hotari <lh...@apache.org>
---
bouncy-castle/bc/pom.xml | 2 +-
bouncy-castle/bcfips-include-test/pom.xml | 2 +-
bouncy-castle/bcfips/pom.xml | 2 +-
bouncy-castle/pom.xml | 2 +-
buildtools/pom.xml | 4 ++--
distribution/io/pom.xml | 2 +-
distribution/offloaders/pom.xml | 2 +-
distribution/pom.xml | 2 +-
distribution/server/pom.xml | 2 +-
distribution/shell/pom.xml | 2 +-
docker/pom.xml | 2 +-
docker/pulsar-all/pom.xml | 2 +-
docker/pulsar/pom.xml | 2 +-
jclouds-shaded/pom.xml | 2 +-
jetcd-core-shaded/pom.xml | 2 +-
managed-ledger/pom.xml | 2 +-
microbench/pom.xml | 2 +-
pom.xml | 4 ++--
pulsar-bom/pom.xml | 4 ++--
pulsar-broker-auth-athenz/pom.xml | 2 +-
pulsar-broker-auth-oidc/pom.xml | 2 +-
pulsar-broker-auth-sasl/pom.xml | 2 +-
pulsar-broker-common/pom.xml | 2 +-
pulsar-broker/pom.xml | 2 +-
pulsar-cli-utils/pom.xml | 2 +-
pulsar-client-admin-api/pom.xml | 2 +-
pulsar-client-admin-shaded/pom.xml | 2 +-
pulsar-client-admin/pom.xml | 2 +-
pulsar-client-all/pom.xml | 2 +-
pulsar-client-api/pom.xml | 2 +-
pulsar-client-auth-athenz/pom.xml | 2 +-
pulsar-client-auth-sasl/pom.xml | 2 +-
pulsar-client-dependencies-minimized/pom.xml | 2 +-
pulsar-client-messagecrypto-bc/pom.xml | 2 +-
pulsar-client-shaded/pom.xml | 2 +-
pulsar-client-tools-api/pom.xml | 2 +-
pulsar-client-tools-customcommand-example/pom.xml | 2 +-
pulsar-client-tools-test/pom.xml | 2 +-
pulsar-client-tools/pom.xml | 2 +-
pulsar-client/pom.xml | 2 +-
pulsar-common/pom.xml | 2 +-
pulsar-config-validation/pom.xml | 2 +-
pulsar-docs-tools/pom.xml | 2 +-
pulsar-functions/api-java/pom.xml | 2 +-
pulsar-functions/instance/pom.xml | 2 +-
pulsar-functions/java-examples-builtin/pom.xml | 2 +-
pulsar-functions/java-examples/pom.xml | 2 +-
pulsar-functions/localrun-shaded/pom.xml | 2 +-
pulsar-functions/localrun/pom.xml | 2 +-
pulsar-functions/pom.xml | 2 +-
pulsar-functions/proto/pom.xml | 2 +-
pulsar-functions/runtime-all/pom.xml | 2 +-
pulsar-functions/runtime/pom.xml | 2 +-
pulsar-functions/secrets/pom.xml | 2 +-
pulsar-functions/utils/pom.xml | 2 +-
pulsar-functions/worker/pom.xml | 2 +-
pulsar-io/aerospike/pom.xml | 2 +-
pulsar-io/alluxio/pom.xml | 2 +-
pulsar-io/aws/pom.xml | 2 +-
pulsar-io/azure-data-explorer/pom.xml | 2 +-
pulsar-io/batch-data-generator/pom.xml | 2 +-
pulsar-io/batch-discovery-triggerers/pom.xml | 2 +-
pulsar-io/canal/pom.xml | 2 +-
pulsar-io/cassandra/pom.xml | 2 +-
pulsar-io/common/pom.xml | 2 +-
pulsar-io/core/pom.xml | 2 +-
pulsar-io/data-generator/pom.xml | 2 +-
pulsar-io/debezium/core/pom.xml | 2 +-
pulsar-io/debezium/mongodb/pom.xml | 2 +-
pulsar-io/debezium/mssql/pom.xml | 2 +-
pulsar-io/debezium/mysql/pom.xml | 2 +-
pulsar-io/debezium/oracle/pom.xml | 2 +-
pulsar-io/debezium/pom.xml | 2 +-
pulsar-io/debezium/postgres/pom.xml | 2 +-
pulsar-io/docs/pom.xml | 2 +-
pulsar-io/dynamodb/pom.xml | 2 +-
pulsar-io/elastic-search/pom.xml | 2 +-
pulsar-io/file/pom.xml | 2 +-
pulsar-io/flume/pom.xml | 2 +-
pulsar-io/hbase/pom.xml | 2 +-
pulsar-io/hdfs3/pom.xml | 2 +-
pulsar-io/http/pom.xml | 2 +-
pulsar-io/influxdb/pom.xml | 2 +-
pulsar-io/jdbc/clickhouse/pom.xml | 2 +-
pulsar-io/jdbc/core/pom.xml | 2 +-
pulsar-io/jdbc/mariadb/pom.xml | 2 +-
pulsar-io/jdbc/openmldb/pom.xml | 2 +-
pulsar-io/jdbc/pom.xml | 2 +-
pulsar-io/jdbc/postgres/pom.xml | 2 +-
pulsar-io/jdbc/sqlite/pom.xml | 2 +-
pulsar-io/kafka-connect-adaptor-nar/pom.xml | 2 +-
pulsar-io/kafka-connect-adaptor/pom.xml | 2 +-
pulsar-io/kafka/pom.xml | 2 +-
pulsar-io/kinesis-kpl-shaded/pom.xml | 2 +-
pulsar-io/kinesis/pom.xml | 2 +-
pulsar-io/mongo/pom.xml | 2 +-
pulsar-io/netty/pom.xml | 2 +-
pulsar-io/nsq/pom.xml | 2 +-
pulsar-io/pom.xml | 2 +-
pulsar-io/rabbitmq/pom.xml | 2 +-
pulsar-io/redis/pom.xml | 2 +-
pulsar-io/solr/pom.xml | 2 +-
pulsar-io/twitter/pom.xml | 2 +-
pulsar-metadata/pom.xml | 2 +-
pulsar-opentelemetry/pom.xml | 2 +-
pulsar-package-management/bookkeeper-storage/pom.xml | 2 +-
pulsar-package-management/core/pom.xml | 2 +-
pulsar-package-management/filesystem-storage/pom.xml | 2 +-
pulsar-package-management/pom.xml | 2 +-
pulsar-proxy/pom.xml | 2 +-
pulsar-testclient/pom.xml | 2 +-
pulsar-transaction/common/pom.xml | 2 +-
pulsar-transaction/coordinator/pom.xml | 2 +-
pulsar-transaction/pom.xml | 2 +-
pulsar-websocket/pom.xml | 2 +-
structured-event-log/pom.xml | 2 +-
testmocks/pom.xml | 2 +-
tests/bc_2_0_0/pom.xml | 2 +-
tests/bc_2_0_1/pom.xml | 2 +-
tests/bc_2_6_0/pom.xml | 2 +-
tests/docker-images/java-test-functions/pom.xml | 2 +-
tests/docker-images/java-test-image/pom.xml | 2 +-
tests/docker-images/java-test-plugins/pom.xml | 2 +-
tests/docker-images/latest-version-image/pom.xml | 2 +-
tests/docker-images/pom.xml | 2 +-
tests/integration/pom.xml | 2 +-
tests/pom.xml | 2 +-
tests/pulsar-client-admin-shade-test/pom.xml | 2 +-
tests/pulsar-client-all-shade-test/pom.xml | 2 +-
tests/pulsar-client-shade-test/pom.xml | 2 +-
tiered-storage/file-system/pom.xml | 2 +-
tiered-storage/jcloud/pom.xml | 2 +-
tiered-storage/pom.xml | 2 +-
133 files changed, 136 insertions(+), 136 deletions(-)
diff --git a/bouncy-castle/bc/pom.xml b/bouncy-castle/bc/pom.xml
index 8a536f07579..6100a0cbba4 100644
--- a/bouncy-castle/bc/pom.xml
+++ b/bouncy-castle/bc/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>bouncy-castle-parent</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>bouncy-castle-bc</artifactId>
diff --git a/bouncy-castle/bcfips-include-test/pom.xml b/bouncy-castle/bcfips-include-test/pom.xml
index 103d5700861..799caee7119 100644
--- a/bouncy-castle/bcfips-include-test/pom.xml
+++ b/bouncy-castle/bcfips-include-test/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>bouncy-castle-parent</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>bcfips-include-test</artifactId>
diff --git a/bouncy-castle/bcfips/pom.xml b/bouncy-castle/bcfips/pom.xml
index 59207031588..40534131b42 100644
--- a/bouncy-castle/bcfips/pom.xml
+++ b/bouncy-castle/bcfips/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>bouncy-castle-parent</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>bouncy-castle-bcfips</artifactId>
diff --git a/bouncy-castle/pom.xml b/bouncy-castle/pom.xml
index 7797acb69ed..36aef1ddd05 100644
--- a/bouncy-castle/pom.xml
+++ b/bouncy-castle/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<build>
diff --git a/buildtools/pom.xml b/buildtools/pom.xml
index 4ffbdfc4a70..de3ff4b8008 100644
--- a/buildtools/pom.xml
+++ b/buildtools/pom.xml
@@ -31,7 +31,7 @@
<groupId>org.apache.pulsar</groupId>
<artifactId>buildtools</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
<packaging>jar</packaging>
<name>Pulsar Build Tools</name>
@@ -43,7 +43,7 @@
</developers>
<properties>
- <project.build.outputTimestamp>2026-02-09T13:43:13Z</project.build.outputTimestamp>
+ <project.build.outputTimestamp>2026-04-15T16:15:41Z</project.build.outputTimestamp>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
<maven.compiler.release>8</maven.compiler.release>
diff --git a/distribution/io/pom.xml b/distribution/io/pom.xml
index 6a8827ace6b..a50cbbb7aab 100644
--- a/distribution/io/pom.xml
+++ b/distribution/io/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>distribution</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-distribution</artifactId>
diff --git a/distribution/offloaders/pom.xml b/distribution/offloaders/pom.xml
index 66af53eaba5..6081fa87455 100644
--- a/distribution/offloaders/pom.xml
+++ b/distribution/offloaders/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>distribution</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-offloader-distribution</artifactId>
diff --git a/distribution/pom.xml b/distribution/pom.xml
index ddc0e02c90d..c9fbd904c78 100644
--- a/distribution/pom.xml
+++ b/distribution/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>distribution</artifactId>
diff --git a/distribution/server/pom.xml b/distribution/server/pom.xml
index 612c4a26814..f2caf97cc95 100644
--- a/distribution/server/pom.xml
+++ b/distribution/server/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>distribution</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-server-distribution</artifactId>
diff --git a/distribution/shell/pom.xml b/distribution/shell/pom.xml
index acc45635d3d..a5f2875448b 100644
--- a/distribution/shell/pom.xml
+++ b/distribution/shell/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>distribution</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-shell-distribution</artifactId>
diff --git a/docker/pom.xml b/docker/pom.xml
index 6bc95cbd76e..38cac854615 100644
--- a/docker/pom.xml
+++ b/docker/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>docker-images</artifactId>
<name>Apache Pulsar :: Docker Images</name>
diff --git a/docker/pulsar-all/pom.xml b/docker/pulsar-all/pom.xml
index e09f9acc0ba..79b88803ca0 100644
--- a/docker/pulsar-all/pom.xml
+++ b/docker/pulsar-all/pom.xml
@@ -23,7 +23,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>docker-images</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>pulsar-all-docker-image</artifactId>
diff --git a/docker/pulsar/pom.xml b/docker/pulsar/pom.xml
index 7475536336d..c9ff353f1b3 100644
--- a/docker/pulsar/pom.xml
+++ b/docker/pulsar/pom.xml
@@ -23,7 +23,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>docker-images</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>pulsar-docker-image</artifactId>
diff --git a/jclouds-shaded/pom.xml b/jclouds-shaded/pom.xml
index 6bc435ac0d7..76dbc7a1ca6 100644
--- a/jclouds-shaded/pom.xml
+++ b/jclouds-shaded/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>jclouds-shaded</artifactId>
diff --git a/jetcd-core-shaded/pom.xml b/jetcd-core-shaded/pom.xml
index 2f012cfcfe8..bd18e688536 100644
--- a/jetcd-core-shaded/pom.xml
+++ b/jetcd-core-shaded/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>jetcd-core-shaded</artifactId>
diff --git a/managed-ledger/pom.xml b/managed-ledger/pom.xml
index 0ad488b53e1..eac0882acda 100644
--- a/managed-ledger/pom.xml
+++ b/managed-ledger/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>managed-ledger</artifactId>
diff --git a/microbench/pom.xml b/microbench/pom.xml
index ddfddec0631..8d0bc6ea6e3 100644
--- a/microbench/pom.xml
+++ b/microbench/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>microbench</artifactId>
diff --git a/pom.xml b/pom.xml
index c88cdcec6ff..4a858b0a290 100644
--- a/pom.xml
+++ b/pom.xml
@@ -32,7 +32,7 @@
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
<name>Pulsar</name>
<description>Pulsar is a distributed pub-sub messaging platform with a very
@@ -94,7 +94,7 @@ flexible messaging model and an intuitive client API.</description>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
- <project.build.outputTimestamp>2026-02-09T13:43:08Z</project.build.outputTimestamp>
+ <project.build.outputTimestamp>2026-04-15T16:15:37Z</project.build.outputTimestamp>
<redirectTestOutputToFile>true</redirectTestOutputToFile>
<!-- By default build for local Docker platform.
diff --git a/pulsar-bom/pom.xml b/pulsar-bom/pom.xml
index 92ca23c83e6..479e557bc59 100644
--- a/pulsar-bom/pom.xml
+++ b/pulsar-bom/pom.xml
@@ -33,7 +33,7 @@
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-bom</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
<name>Pulsar BOM</name>
<description>Pulsar (Bill of Materials)</description>
@@ -81,7 +81,7 @@
<maven.compiler.target>17</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
- <project.build.outputTimestamp>2026-02-09T13:43:16Z</project.build.outputTimestamp>
+ <project.build.outputTimestamp>2026-04-15T16:15:43Z</project.build.outputTimestamp>
<license-maven-plugin.version>4.1</license-maven-plugin.version>
<maven-checkstyle-plugin.version>3.1.2</maven-checkstyle-plugin.version>
<wagon-ssh-external.version>3.5.3</wagon-ssh-external.version>
diff --git a/pulsar-broker-auth-athenz/pom.xml b/pulsar-broker-auth-athenz/pom.xml
index 23d68bac027..2eb0a52d788 100644
--- a/pulsar-broker-auth-athenz/pom.xml
+++ b/pulsar-broker-auth-athenz/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-broker-auth-athenz</artifactId>
diff --git a/pulsar-broker-auth-oidc/pom.xml b/pulsar-broker-auth-oidc/pom.xml
index 44e9e120c34..08a5fa95de8 100644
--- a/pulsar-broker-auth-oidc/pom.xml
+++ b/pulsar-broker-auth-oidc/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-broker-auth-oidc</artifactId>
diff --git a/pulsar-broker-auth-sasl/pom.xml b/pulsar-broker-auth-sasl/pom.xml
index 1d7c67a89a1..8ed91a4e7d8 100644
--- a/pulsar-broker-auth-sasl/pom.xml
+++ b/pulsar-broker-auth-sasl/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-broker-auth-sasl</artifactId>
diff --git a/pulsar-broker-common/pom.xml b/pulsar-broker-common/pom.xml
index 9a5f9f756b5..82484191a43 100644
--- a/pulsar-broker-common/pom.xml
+++ b/pulsar-broker-common/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-broker-common</artifactId>
diff --git a/pulsar-broker/pom.xml b/pulsar-broker/pom.xml
index 6f3e65ead1d..5fe1d553eb4 100644
--- a/pulsar-broker/pom.xml
+++ b/pulsar-broker/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-broker</artifactId>
diff --git a/pulsar-cli-utils/pom.xml b/pulsar-cli-utils/pom.xml
index 50232b5017c..48e0f008e46 100644
--- a/pulsar-cli-utils/pom.xml
+++ b/pulsar-cli-utils/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-cli-utils</artifactId>
diff --git a/pulsar-client-admin-api/pom.xml b/pulsar-client-admin-api/pom.xml
index ac7f6f14803..95b14638459 100644
--- a/pulsar-client-admin-api/pom.xml
+++ b/pulsar-client-admin-api/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-client-admin-api</artifactId>
diff --git a/pulsar-client-admin-shaded/pom.xml b/pulsar-client-admin-shaded/pom.xml
index 2ac6ca47325..d8d1b83a1b5 100644
--- a/pulsar-client-admin-shaded/pom.xml
+++ b/pulsar-client-admin-shaded/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-client-admin</artifactId>
<name>Pulsar Client Admin</name>
diff --git a/pulsar-client-admin/pom.xml b/pulsar-client-admin/pom.xml
index cc91b9683de..8c4fa40016a 100644
--- a/pulsar-client-admin/pom.xml
+++ b/pulsar-client-admin/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-client-admin-original</artifactId>
diff --git a/pulsar-client-all/pom.xml b/pulsar-client-all/pom.xml
index e3bf3a3120b..f75ce52e1f0 100644
--- a/pulsar-client-all/pom.xml
+++ b/pulsar-client-all/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-client-all</artifactId>
<name>Pulsar Client All</name>
diff --git a/pulsar-client-api/pom.xml b/pulsar-client-api/pom.xml
index 46879df7acb..dcfa0690bac 100644
--- a/pulsar-client-api/pom.xml
+++ b/pulsar-client-api/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-client-api</artifactId>
diff --git a/pulsar-client-auth-athenz/pom.xml b/pulsar-client-auth-athenz/pom.xml
index 8190fd88ada..e7378aec7bb 100644
--- a/pulsar-client-auth-athenz/pom.xml
+++ b/pulsar-client-auth-athenz/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-client-auth-athenz</artifactId>
diff --git a/pulsar-client-auth-sasl/pom.xml b/pulsar-client-auth-sasl/pom.xml
index 1b7ecd5148e..cf9def24835 100644
--- a/pulsar-client-auth-sasl/pom.xml
+++ b/pulsar-client-auth-sasl/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-client-auth-sasl</artifactId>
diff --git a/pulsar-client-dependencies-minimized/pom.xml b/pulsar-client-dependencies-minimized/pom.xml
index 73252b4a93c..d19c0834ea0 100644
--- a/pulsar-client-dependencies-minimized/pom.xml
+++ b/pulsar-client-dependencies-minimized/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-client-dependencies-minimized</artifactId>
diff --git a/pulsar-client-messagecrypto-bc/pom.xml b/pulsar-client-messagecrypto-bc/pom.xml
index c5db9d93b88..baa4d9136c6 100644
--- a/pulsar-client-messagecrypto-bc/pom.xml
+++ b/pulsar-client-messagecrypto-bc/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-client-messagecrypto-bc</artifactId>
diff --git a/pulsar-client-shaded/pom.xml b/pulsar-client-shaded/pom.xml
index fd0526963dc..40f89a6aff9 100644
--- a/pulsar-client-shaded/pom.xml
+++ b/pulsar-client-shaded/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-client</artifactId>
<name>Pulsar Client Java</name>
diff --git a/pulsar-client-tools-api/pom.xml b/pulsar-client-tools-api/pom.xml
index 5e4cd4cb7d6..ce92726a636 100644
--- a/pulsar-client-tools-api/pom.xml
+++ b/pulsar-client-tools-api/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-client-tools-api</artifactId>
diff --git a/pulsar-client-tools-customcommand-example/pom.xml b/pulsar-client-tools-customcommand-example/pom.xml
index dcea02f12f0..662e6e8959a 100644
--- a/pulsar-client-tools-customcommand-example/pom.xml
+++ b/pulsar-client-tools-customcommand-example/pom.xml
@@ -22,7 +22,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>pulsar-client-tools-customcommand-example</artifactId>
diff --git a/pulsar-client-tools-test/pom.xml b/pulsar-client-tools-test/pom.xml
index ba5aa0602a5..725b3c36e5a 100644
--- a/pulsar-client-tools-test/pom.xml
+++ b/pulsar-client-tools-test/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-client-tools-test</artifactId>
diff --git a/pulsar-client-tools/pom.xml b/pulsar-client-tools/pom.xml
index 5c4e3f13693..32374a2e412 100644
--- a/pulsar-client-tools/pom.xml
+++ b/pulsar-client-tools/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-client-tools</artifactId>
diff --git a/pulsar-client/pom.xml b/pulsar-client/pom.xml
index 7c5eef8986e..220d7630513 100644
--- a/pulsar-client/pom.xml
+++ b/pulsar-client/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-client-original</artifactId>
diff --git a/pulsar-common/pom.xml b/pulsar-common/pom.xml
index e7339e4e31b..c6c2fa4cf3c 100644
--- a/pulsar-common/pom.xml
+++ b/pulsar-common/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-common</artifactId>
diff --git a/pulsar-config-validation/pom.xml b/pulsar-config-validation/pom.xml
index 3e2b0c8c27d..56666c1fce7 100644
--- a/pulsar-config-validation/pom.xml
+++ b/pulsar-config-validation/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-config-validation</artifactId>
diff --git a/pulsar-docs-tools/pom.xml b/pulsar-docs-tools/pom.xml
index 0894d98abf4..b9ad36e97cf 100644
--- a/pulsar-docs-tools/pom.xml
+++ b/pulsar-docs-tools/pom.xml
@@ -27,7 +27,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-docs-tools</artifactId>
diff --git a/pulsar-functions/api-java/pom.xml b/pulsar-functions/api-java/pom.xml
index ee0b1360691..4c79a77ca5e 100644
--- a/pulsar-functions/api-java/pom.xml
+++ b/pulsar-functions/api-java/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-functions</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-functions-api</artifactId>
diff --git a/pulsar-functions/instance/pom.xml b/pulsar-functions/instance/pom.xml
index c2e0a94aabc..64b8e39e636 100644
--- a/pulsar-functions/instance/pom.xml
+++ b/pulsar-functions/instance/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-functions</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-functions-instance</artifactId>
diff --git a/pulsar-functions/java-examples-builtin/pom.xml b/pulsar-functions/java-examples-builtin/pom.xml
index 730ce2f9eea..af354037d1d 100644
--- a/pulsar-functions/java-examples-builtin/pom.xml
+++ b/pulsar-functions/java-examples-builtin/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-functions</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-functions-api-examples-builtin</artifactId>
diff --git a/pulsar-functions/java-examples/pom.xml b/pulsar-functions/java-examples/pom.xml
index 58e10d7cf94..d74824f8324 100644
--- a/pulsar-functions/java-examples/pom.xml
+++ b/pulsar-functions/java-examples/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-functions</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-functions-api-examples</artifactId>
diff --git a/pulsar-functions/localrun-shaded/pom.xml b/pulsar-functions/localrun-shaded/pom.xml
index fa6b9345c5e..b1937f75d51 100644
--- a/pulsar-functions/localrun-shaded/pom.xml
+++ b/pulsar-functions/localrun-shaded/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-functions</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-functions-local-runner</artifactId>
diff --git a/pulsar-functions/localrun/pom.xml b/pulsar-functions/localrun/pom.xml
index 9d1037c4b2e..82b544f7b74 100644
--- a/pulsar-functions/localrun/pom.xml
+++ b/pulsar-functions/localrun/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-functions</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-functions-local-runner-original</artifactId>
diff --git a/pulsar-functions/pom.xml b/pulsar-functions/pom.xml
index 0e67b259faa..714757fb511 100644
--- a/pulsar-functions/pom.xml
+++ b/pulsar-functions/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-functions</artifactId>
diff --git a/pulsar-functions/proto/pom.xml b/pulsar-functions/proto/pom.xml
index 72a68b97b51..fb83d4973da 100644
--- a/pulsar-functions/proto/pom.xml
+++ b/pulsar-functions/proto/pom.xml
@@ -27,7 +27,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-functions</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-functions-proto</artifactId>
diff --git a/pulsar-functions/runtime-all/pom.xml b/pulsar-functions/runtime-all/pom.xml
index 9919423739c..a6890fa182b 100644
--- a/pulsar-functions/runtime-all/pom.xml
+++ b/pulsar-functions/runtime-all/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-functions</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<!--
diff --git a/pulsar-functions/runtime/pom.xml b/pulsar-functions/runtime/pom.xml
index 6b8df639ecf..5d8752a73a2 100644
--- a/pulsar-functions/runtime/pom.xml
+++ b/pulsar-functions/runtime/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-functions</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-functions-runtime</artifactId>
diff --git a/pulsar-functions/secrets/pom.xml b/pulsar-functions/secrets/pom.xml
index f6d355ba66f..7198d43414c 100644
--- a/pulsar-functions/secrets/pom.xml
+++ b/pulsar-functions/secrets/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-functions</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-functions-secrets</artifactId>
diff --git a/pulsar-functions/utils/pom.xml b/pulsar-functions/utils/pom.xml
index 14184469ba9..3c94103822c 100644
--- a/pulsar-functions/utils/pom.xml
+++ b/pulsar-functions/utils/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-functions</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-functions-utils</artifactId>
diff --git a/pulsar-functions/worker/pom.xml b/pulsar-functions/worker/pom.xml
index b0d2efa5923..6ac81c0c2f4 100644
--- a/pulsar-functions/worker/pom.xml
+++ b/pulsar-functions/worker/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-functions</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-functions-worker</artifactId>
diff --git a/pulsar-io/aerospike/pom.xml b/pulsar-io/aerospike/pom.xml
index de7a795df8c..b0ac0a2ae47 100644
--- a/pulsar-io/aerospike/pom.xml
+++ b/pulsar-io/aerospike/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-aerospike</artifactId>
diff --git a/pulsar-io/alluxio/pom.xml b/pulsar-io/alluxio/pom.xml
index 29d0a398722..58424962d69 100644
--- a/pulsar-io/alluxio/pom.xml
+++ b/pulsar-io/alluxio/pom.xml
@@ -25,7 +25,7 @@
<parent>
<artifactId>pulsar-io</artifactId>
<groupId>org.apache.pulsar</groupId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<properties>
diff --git a/pulsar-io/aws/pom.xml b/pulsar-io/aws/pom.xml
index e64167ab350..5efc61c6fad 100644
--- a/pulsar-io/aws/pom.xml
+++ b/pulsar-io/aws/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-aws</artifactId>
diff --git a/pulsar-io/azure-data-explorer/pom.xml b/pulsar-io/azure-data-explorer/pom.xml
index 98b6ffa1643..e9b8ebf3a63 100644
--- a/pulsar-io/azure-data-explorer/pom.xml
+++ b/pulsar-io/azure-data-explorer/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-azuredataexplorer</artifactId>
diff --git a/pulsar-io/batch-data-generator/pom.xml b/pulsar-io/batch-data-generator/pom.xml
index 3b690810f97..5b775146185 100644
--- a/pulsar-io/batch-data-generator/pom.xml
+++ b/pulsar-io/batch-data-generator/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-batch-data-generator</artifactId>
diff --git a/pulsar-io/batch-discovery-triggerers/pom.xml b/pulsar-io/batch-discovery-triggerers/pom.xml
index b9affdb5145..a819a516bbf 100644
--- a/pulsar-io/batch-discovery-triggerers/pom.xml
+++ b/pulsar-io/batch-discovery-triggerers/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-batch-discovery-triggerers</artifactId>
diff --git a/pulsar-io/canal/pom.xml b/pulsar-io/canal/pom.xml
index 3bee821ca58..98177186a17 100644
--- a/pulsar-io/canal/pom.xml
+++ b/pulsar-io/canal/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pulsar-io/cassandra/pom.xml b/pulsar-io/cassandra/pom.xml
index bb233b1d920..9c0354fb57f 100644
--- a/pulsar-io/cassandra/pom.xml
+++ b/pulsar-io/cassandra/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-cassandra</artifactId>
diff --git a/pulsar-io/common/pom.xml b/pulsar-io/common/pom.xml
index feb457aeb5b..cadc8801fd5 100644
--- a/pulsar-io/common/pom.xml
+++ b/pulsar-io/common/pom.xml
@@ -27,7 +27,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-common</artifactId>
diff --git a/pulsar-io/core/pom.xml b/pulsar-io/core/pom.xml
index 9d7a4f37594..b2d54566221 100644
--- a/pulsar-io/core/pom.xml
+++ b/pulsar-io/core/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-core</artifactId>
diff --git a/pulsar-io/data-generator/pom.xml b/pulsar-io/data-generator/pom.xml
index fb43d78ff7b..860276d312a 100644
--- a/pulsar-io/data-generator/pom.xml
+++ b/pulsar-io/data-generator/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-data-generator</artifactId>
diff --git a/pulsar-io/debezium/core/pom.xml b/pulsar-io/debezium/core/pom.xml
index 193d9d6cddf..dbfac00fb95 100644
--- a/pulsar-io/debezium/core/pom.xml
+++ b/pulsar-io/debezium/core/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io-debezium</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-debezium-core</artifactId>
diff --git a/pulsar-io/debezium/mongodb/pom.xml b/pulsar-io/debezium/mongodb/pom.xml
index 73b925bf964..4f034e65a78 100644
--- a/pulsar-io/debezium/mongodb/pom.xml
+++ b/pulsar-io/debezium/mongodb/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io-debezium</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-debezium-mongodb</artifactId>
diff --git a/pulsar-io/debezium/mssql/pom.xml b/pulsar-io/debezium/mssql/pom.xml
index 4ebcfb081a5..b7a9d4a3591 100644
--- a/pulsar-io/debezium/mssql/pom.xml
+++ b/pulsar-io/debezium/mssql/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io-debezium</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-debezium-mssql</artifactId>
diff --git a/pulsar-io/debezium/mysql/pom.xml b/pulsar-io/debezium/mysql/pom.xml
index 011a39112e8..68fdd0c8338 100644
--- a/pulsar-io/debezium/mysql/pom.xml
+++ b/pulsar-io/debezium/mysql/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io-debezium</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-debezium-mysql</artifactId>
diff --git a/pulsar-io/debezium/oracle/pom.xml b/pulsar-io/debezium/oracle/pom.xml
index fb34115ba27..c26d97fd09d 100644
--- a/pulsar-io/debezium/oracle/pom.xml
+++ b/pulsar-io/debezium/oracle/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io-debezium</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-debezium-oracle</artifactId>
diff --git a/pulsar-io/debezium/pom.xml b/pulsar-io/debezium/pom.xml
index bc8c15b98a9..1c650791a08 100644
--- a/pulsar-io/debezium/pom.xml
+++ b/pulsar-io/debezium/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-debezium</artifactId>
diff --git a/pulsar-io/debezium/postgres/pom.xml b/pulsar-io/debezium/postgres/pom.xml
index 76565b876e3..2ef3f91332a 100644
--- a/pulsar-io/debezium/postgres/pom.xml
+++ b/pulsar-io/debezium/postgres/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io-debezium</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-debezium-postgres</artifactId>
diff --git a/pulsar-io/docs/pom.xml b/pulsar-io/docs/pom.xml
index 16d7d1f68ad..d76ddb0b71c 100644
--- a/pulsar-io/docs/pom.xml
+++ b/pulsar-io/docs/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-docs</artifactId>
diff --git a/pulsar-io/dynamodb/pom.xml b/pulsar-io/dynamodb/pom.xml
index 7356111ec01..beb9fcfc1bc 100644
--- a/pulsar-io/dynamodb/pom.xml
+++ b/pulsar-io/dynamodb/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-dynamodb</artifactId>
diff --git a/pulsar-io/elastic-search/pom.xml b/pulsar-io/elastic-search/pom.xml
index 14bea2badd1..aea7f8506b1 100644
--- a/pulsar-io/elastic-search/pom.xml
+++ b/pulsar-io/elastic-search/pom.xml
@@ -23,7 +23,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-elastic-search</artifactId>
<name>Pulsar IO :: ElasticSearch</name>
diff --git a/pulsar-io/file/pom.xml b/pulsar-io/file/pom.xml
index ce3d253d660..263c70f34ca 100644
--- a/pulsar-io/file/pom.xml
+++ b/pulsar-io/file/pom.xml
@@ -23,7 +23,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-file</artifactId>
diff --git a/pulsar-io/flume/pom.xml b/pulsar-io/flume/pom.xml
index 5b3dd9686dd..11bdcd5a1ed 100644
--- a/pulsar-io/flume/pom.xml
+++ b/pulsar-io/flume/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-flume</artifactId>
diff --git a/pulsar-io/hbase/pom.xml b/pulsar-io/hbase/pom.xml
index 1b3e0d0c87c..2d028f468e7 100644
--- a/pulsar-io/hbase/pom.xml
+++ b/pulsar-io/hbase/pom.xml
@@ -25,7 +25,7 @@
<parent>
<artifactId>pulsar-io</artifactId>
<groupId>org.apache.pulsar</groupId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-hbase</artifactId>
<name>Pulsar IO :: Hbase</name>
diff --git a/pulsar-io/hdfs3/pom.xml b/pulsar-io/hdfs3/pom.xml
index 69dd7e6e533..23508590a84 100644
--- a/pulsar-io/hdfs3/pom.xml
+++ b/pulsar-io/hdfs3/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-hdfs3</artifactId>
<name>Pulsar IO :: Hdfs3</name>
diff --git a/pulsar-io/http/pom.xml b/pulsar-io/http/pom.xml
index 8994890031d..8c224812a7b 100644
--- a/pulsar-io/http/pom.xml
+++ b/pulsar-io/http/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-http</artifactId>
diff --git a/pulsar-io/influxdb/pom.xml b/pulsar-io/influxdb/pom.xml
index e4228b54b8e..0e5fa09a893 100644
--- a/pulsar-io/influxdb/pom.xml
+++ b/pulsar-io/influxdb/pom.xml
@@ -25,7 +25,7 @@
<parent>
<artifactId>pulsar-io</artifactId>
<groupId>org.apache.pulsar</groupId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-influxdb</artifactId>
diff --git a/pulsar-io/jdbc/clickhouse/pom.xml b/pulsar-io/jdbc/clickhouse/pom.xml
index d60912cbcca..c74e6ebb707 100644
--- a/pulsar-io/jdbc/clickhouse/pom.xml
+++ b/pulsar-io/jdbc/clickhouse/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pulsar-io-jdbc</artifactId>
<groupId>org.apache.pulsar</groupId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pulsar-io/jdbc/core/pom.xml b/pulsar-io/jdbc/core/pom.xml
index a76fd9e8610..d32e80dc99c 100644
--- a/pulsar-io/jdbc/core/pom.xml
+++ b/pulsar-io/jdbc/core/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pulsar-io-jdbc</artifactId>
<groupId>org.apache.pulsar</groupId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pulsar-io/jdbc/mariadb/pom.xml b/pulsar-io/jdbc/mariadb/pom.xml
index 2b71f3fbf5b..7be2391ea0f 100644
--- a/pulsar-io/jdbc/mariadb/pom.xml
+++ b/pulsar-io/jdbc/mariadb/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pulsar-io-jdbc</artifactId>
<groupId>org.apache.pulsar</groupId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pulsar-io/jdbc/openmldb/pom.xml b/pulsar-io/jdbc/openmldb/pom.xml
index 1bd80578777..fe06090f171 100644
--- a/pulsar-io/jdbc/openmldb/pom.xml
+++ b/pulsar-io/jdbc/openmldb/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pulsar-io-jdbc</artifactId>
<groupId>org.apache.pulsar</groupId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pulsar-io/jdbc/pom.xml b/pulsar-io/jdbc/pom.xml
index c32410368c7..a48fc6d6503 100644
--- a/pulsar-io/jdbc/pom.xml
+++ b/pulsar-io/jdbc/pom.xml
@@ -33,7 +33,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-jdbc</artifactId>
diff --git a/pulsar-io/jdbc/postgres/pom.xml b/pulsar-io/jdbc/postgres/pom.xml
index 1627a2ca1e4..4ff470c2028 100644
--- a/pulsar-io/jdbc/postgres/pom.xml
+++ b/pulsar-io/jdbc/postgres/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pulsar-io-jdbc</artifactId>
<groupId>org.apache.pulsar</groupId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pulsar-io/jdbc/sqlite/pom.xml b/pulsar-io/jdbc/sqlite/pom.xml
index f7177798d71..625ba75f7c2 100644
--- a/pulsar-io/jdbc/sqlite/pom.xml
+++ b/pulsar-io/jdbc/sqlite/pom.xml
@@ -24,7 +24,7 @@
<parent>
<artifactId>pulsar-io-jdbc</artifactId>
<groupId>org.apache.pulsar</groupId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>pulsar-io-jdbc-sqlite</artifactId>
diff --git a/pulsar-io/kafka-connect-adaptor-nar/pom.xml b/pulsar-io/kafka-connect-adaptor-nar/pom.xml
index 6d158b0648a..b0b384ee7c8 100644
--- a/pulsar-io/kafka-connect-adaptor-nar/pom.xml
+++ b/pulsar-io/kafka-connect-adaptor-nar/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-kafka-connect-adaptor-nar</artifactId>
diff --git a/pulsar-io/kafka-connect-adaptor/pom.xml b/pulsar-io/kafka-connect-adaptor/pom.xml
index 4b12ad71e93..b9c75d200f8 100644
--- a/pulsar-io/kafka-connect-adaptor/pom.xml
+++ b/pulsar-io/kafka-connect-adaptor/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-kafka-connect-adaptor</artifactId>
diff --git a/pulsar-io/kafka/pom.xml b/pulsar-io/kafka/pom.xml
index 0301309aad0..43a1cf3b132 100644
--- a/pulsar-io/kafka/pom.xml
+++ b/pulsar-io/kafka/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-kafka</artifactId>
diff --git a/pulsar-io/kinesis-kpl-shaded/pom.xml b/pulsar-io/kinesis-kpl-shaded/pom.xml
index b6bb13d69a6..61d54c8f2a7 100644
--- a/pulsar-io/kinesis-kpl-shaded/pom.xml
+++ b/pulsar-io/kinesis-kpl-shaded/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
<relativePath>../../pom.xml</relativePath>
</parent>
diff --git a/pulsar-io/kinesis/pom.xml b/pulsar-io/kinesis/pom.xml
index 10a5644581a..9d0b1cc65e3 100644
--- a/pulsar-io/kinesis/pom.xml
+++ b/pulsar-io/kinesis/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-kinesis</artifactId>
diff --git a/pulsar-io/mongo/pom.xml b/pulsar-io/mongo/pom.xml
index 33c0b9ce207..7f79dd66df7 100644
--- a/pulsar-io/mongo/pom.xml
+++ b/pulsar-io/mongo/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-mongo</artifactId>
diff --git a/pulsar-io/netty/pom.xml b/pulsar-io/netty/pom.xml
index f072ba80181..13e7ac67e54 100644
--- a/pulsar-io/netty/pom.xml
+++ b/pulsar-io/netty/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-netty</artifactId>
diff --git a/pulsar-io/nsq/pom.xml b/pulsar-io/nsq/pom.xml
index 0ecb6845634..f2dbb1ae4dd 100644
--- a/pulsar-io/nsq/pom.xml
+++ b/pulsar-io/nsq/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-nsq</artifactId>
diff --git a/pulsar-io/pom.xml b/pulsar-io/pom.xml
index 0d20f0d27ca..a1dae102c81 100644
--- a/pulsar-io/pom.xml
+++ b/pulsar-io/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io</artifactId>
diff --git a/pulsar-io/rabbitmq/pom.xml b/pulsar-io/rabbitmq/pom.xml
index 5cb6a378ef2..7fe2446a030 100644
--- a/pulsar-io/rabbitmq/pom.xml
+++ b/pulsar-io/rabbitmq/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-rabbitmq</artifactId>
diff --git a/pulsar-io/redis/pom.xml b/pulsar-io/redis/pom.xml
index 42deea8ff9b..56e5e2fbf0c 100644
--- a/pulsar-io/redis/pom.xml
+++ b/pulsar-io/redis/pom.xml
@@ -25,7 +25,7 @@
<parent>
<artifactId>pulsar-io</artifactId>
<groupId>org.apache.pulsar</groupId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-redis</artifactId>
diff --git a/pulsar-io/solr/pom.xml b/pulsar-io/solr/pom.xml
index 48ce00506b4..bcbaf06e423 100644
--- a/pulsar-io/solr/pom.xml
+++ b/pulsar-io/solr/pom.xml
@@ -25,7 +25,7 @@
<parent>
<artifactId>pulsar-io</artifactId>
<groupId>org.apache.pulsar</groupId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<properties>
diff --git a/pulsar-io/twitter/pom.xml b/pulsar-io/twitter/pom.xml
index 7153b94dc15..896cc7599d6 100644
--- a/pulsar-io/twitter/pom.xml
+++ b/pulsar-io/twitter/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-io</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-io-twitter</artifactId>
diff --git a/pulsar-metadata/pom.xml b/pulsar-metadata/pom.xml
index c970b111932..1a27b0fc347 100644
--- a/pulsar-metadata/pom.xml
+++ b/pulsar-metadata/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<properties>
diff --git a/pulsar-opentelemetry/pom.xml b/pulsar-opentelemetry/pom.xml
index 3f521bb4127..036aff24d3b 100644
--- a/pulsar-opentelemetry/pom.xml
+++ b/pulsar-opentelemetry/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-opentelemetry</artifactId>
diff --git a/pulsar-package-management/bookkeeper-storage/pom.xml b/pulsar-package-management/bookkeeper-storage/pom.xml
index 9ebc45262c3..ec56e4f09e7 100644
--- a/pulsar-package-management/bookkeeper-storage/pom.xml
+++ b/pulsar-package-management/bookkeeper-storage/pom.xml
@@ -25,7 +25,7 @@
<parent>
<artifactId>pulsar-package-management</artifactId>
<groupId>org.apache.pulsar</groupId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pulsar-package-management/core/pom.xml b/pulsar-package-management/core/pom.xml
index 5ca92cb5ced..49d1439e51d 100644
--- a/pulsar-package-management/core/pom.xml
+++ b/pulsar-package-management/core/pom.xml
@@ -25,7 +25,7 @@
<parent>
<artifactId>pulsar-package-management</artifactId>
<groupId>org.apache.pulsar</groupId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pulsar-package-management/filesystem-storage/pom.xml b/pulsar-package-management/filesystem-storage/pom.xml
index ed4199d25ee..049fd80f2ef 100644
--- a/pulsar-package-management/filesystem-storage/pom.xml
+++ b/pulsar-package-management/filesystem-storage/pom.xml
@@ -25,7 +25,7 @@
<parent>
<artifactId>pulsar-package-management</artifactId>
<groupId>org.apache.pulsar</groupId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pulsar-package-management/pom.xml b/pulsar-package-management/pom.xml
index bb18cb03881..eb56b139d5a 100644
--- a/pulsar-package-management/pom.xml
+++ b/pulsar-package-management/pom.xml
@@ -25,7 +25,7 @@
<parent>
<artifactId>pulsar</artifactId>
<groupId>org.apache.pulsar</groupId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/pulsar-proxy/pom.xml b/pulsar-proxy/pom.xml
index d241c4f13d4..1672353d638 100644
--- a/pulsar-proxy/pom.xml
+++ b/pulsar-proxy/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-proxy</artifactId>
diff --git a/pulsar-testclient/pom.xml b/pulsar-testclient/pom.xml
index 1e319a3b575..7f7c90677ef 100644
--- a/pulsar-testclient/pom.xml
+++ b/pulsar-testclient/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-testclient</artifactId>
diff --git a/pulsar-transaction/common/pom.xml b/pulsar-transaction/common/pom.xml
index e9a974f2b64..c4cb314f03c 100644
--- a/pulsar-transaction/common/pom.xml
+++ b/pulsar-transaction/common/pom.xml
@@ -27,7 +27,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-transaction-parent</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-transaction-common</artifactId>
diff --git a/pulsar-transaction/coordinator/pom.xml b/pulsar-transaction/coordinator/pom.xml
index 12592039007..751a4d739e7 100644
--- a/pulsar-transaction/coordinator/pom.xml
+++ b/pulsar-transaction/coordinator/pom.xml
@@ -27,7 +27,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-transaction-parent</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-transaction-coordinator</artifactId>
diff --git a/pulsar-transaction/pom.xml b/pulsar-transaction/pom.xml
index 60592e326ba..712d5fed8f8 100644
--- a/pulsar-transaction/pom.xml
+++ b/pulsar-transaction/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-transaction-parent</artifactId>
diff --git a/pulsar-websocket/pom.xml b/pulsar-websocket/pom.xml
index fc90de1222d..b8368d745f2 100644
--- a/pulsar-websocket/pom.xml
+++ b/pulsar-websocket/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-websocket</artifactId>
diff --git a/structured-event-log/pom.xml b/structured-event-log/pom.xml
index 82c2148550c..0d7004aa8cb 100644
--- a/structured-event-log/pom.xml
+++ b/structured-event-log/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>structured-event-log</artifactId>
diff --git a/testmocks/pom.xml b/testmocks/pom.xml
index fc6f781b977..505b8cb9609 100644
--- a/testmocks/pom.xml
+++ b/testmocks/pom.xml
@@ -25,7 +25,7 @@
<parent>
<artifactId>pulsar</artifactId>
<groupId>org.apache.pulsar</groupId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>testmocks</artifactId>
diff --git a/tests/bc_2_0_0/pom.xml b/tests/bc_2_0_0/pom.xml
index 6b4bd87990b..c8d3f41a65b 100644
--- a/tests/bc_2_0_0/pom.xml
+++ b/tests/bc_2_0_0/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar.tests</groupId>
<artifactId>tests-parent</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>bc_2_0_0</artifactId>
diff --git a/tests/bc_2_0_1/pom.xml b/tests/bc_2_0_1/pom.xml
index 31a7b565bcc..8d1a2fef5ae 100644
--- a/tests/bc_2_0_1/pom.xml
+++ b/tests/bc_2_0_1/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar.tests</groupId>
<artifactId>tests-parent</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>bc_2_0_1</artifactId>
diff --git a/tests/bc_2_6_0/pom.xml b/tests/bc_2_6_0/pom.xml
index 0543746666b..fb0d2d9909a 100644
--- a/tests/bc_2_6_0/pom.xml
+++ b/tests/bc_2_6_0/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar.tests</groupId>
<artifactId>tests-parent</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/tests/docker-images/java-test-functions/pom.xml b/tests/docker-images/java-test-functions/pom.xml
index 36ef28a9ad3..eb4b9cf8194 100644
--- a/tests/docker-images/java-test-functions/pom.xml
+++ b/tests/docker-images/java-test-functions/pom.xml
@@ -23,7 +23,7 @@
<parent>
<groupId>org.apache.pulsar.tests</groupId>
<artifactId>docker-images</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>java-test-functions</artifactId>
diff --git a/tests/docker-images/java-test-image/pom.xml b/tests/docker-images/java-test-image/pom.xml
index 04087f9cccb..6589a4e5129 100644
--- a/tests/docker-images/java-test-image/pom.xml
+++ b/tests/docker-images/java-test-image/pom.xml
@@ -23,7 +23,7 @@
<parent>
<groupId>org.apache.pulsar.tests</groupId>
<artifactId>docker-images</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>java-test-image</artifactId>
diff --git a/tests/docker-images/java-test-plugins/pom.xml b/tests/docker-images/java-test-plugins/pom.xml
index e3dff2da965..aa831e487b3 100644
--- a/tests/docker-images/java-test-plugins/pom.xml
+++ b/tests/docker-images/java-test-plugins/pom.xml
@@ -23,7 +23,7 @@
<parent>
<groupId>org.apache.pulsar.tests</groupId>
<artifactId>docker-images</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>java-test-plugins</artifactId>
diff --git a/tests/docker-images/latest-version-image/pom.xml b/tests/docker-images/latest-version-image/pom.xml
index 94eff8caceb..bebd1976a33 100644
--- a/tests/docker-images/latest-version-image/pom.xml
+++ b/tests/docker-images/latest-version-image/pom.xml
@@ -23,7 +23,7 @@
<parent>
<groupId>org.apache.pulsar.tests</groupId>
<artifactId>docker-images</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>latest-version-image</artifactId>
diff --git a/tests/docker-images/pom.xml b/tests/docker-images/pom.xml
index 8316c06742d..f775dd9b4c1 100644
--- a/tests/docker-images/pom.xml
+++ b/tests/docker-images/pom.xml
@@ -27,7 +27,7 @@
<parent>
<groupId>org.apache.pulsar.tests</groupId>
<artifactId>tests-parent</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>docker-images</artifactId>
<name>Apache Pulsar :: Tests :: Docker Images</name>
diff --git a/tests/integration/pom.xml b/tests/integration/pom.xml
index 8322de14366..75213fb73b4 100644
--- a/tests/integration/pom.xml
+++ b/tests/integration/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar.tests</groupId>
<artifactId>tests-parent</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>integration</artifactId>
diff --git a/tests/pom.xml b/tests/pom.xml
index 37052af1202..63ce932a13d 100644
--- a/tests/pom.xml
+++ b/tests/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<groupId>org.apache.pulsar.tests</groupId>
<artifactId>tests-parent</artifactId>
diff --git a/tests/pulsar-client-admin-shade-test/pom.xml b/tests/pulsar-client-admin-shade-test/pom.xml
index 5b53415d6b5..4dfb7dfecc6 100644
--- a/tests/pulsar-client-admin-shade-test/pom.xml
+++ b/tests/pulsar-client-admin-shade-test/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar.tests</groupId>
<artifactId>tests-parent</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-client-admin-shade-test</artifactId>
diff --git a/tests/pulsar-client-all-shade-test/pom.xml b/tests/pulsar-client-all-shade-test/pom.xml
index 1416b21f82f..6748ab1093e 100644
--- a/tests/pulsar-client-all-shade-test/pom.xml
+++ b/tests/pulsar-client-all-shade-test/pom.xml
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.pulsar.tests</groupId>
<artifactId>tests-parent</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-client-all-shade-test</artifactId>
diff --git a/tests/pulsar-client-shade-test/pom.xml b/tests/pulsar-client-shade-test/pom.xml
index 1548e4c492f..e7090ad8bea 100644
--- a/tests/pulsar-client-shade-test/pom.xml
+++ b/tests/pulsar-client-shade-test/pom.xml
@@ -27,7 +27,7 @@
<parent>
<groupId>org.apache.pulsar.tests</groupId>
<artifactId>tests-parent</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>pulsar-client-shade-test</artifactId>
diff --git a/tiered-storage/file-system/pom.xml b/tiered-storage/file-system/pom.xml
index b03a05198ad..8153456fa0a 100644
--- a/tiered-storage/file-system/pom.xml
+++ b/tiered-storage/file-system/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>tiered-storage-parent</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>tiered-storage-file-system</artifactId>
diff --git a/tiered-storage/jcloud/pom.xml b/tiered-storage/jcloud/pom.xml
index c330e33b3ae..c16a7157d2d 100644
--- a/tiered-storage/jcloud/pom.xml
+++ b/tiered-storage/jcloud/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>tiered-storage-parent</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>tiered-storage-jcloud</artifactId>
diff --git a/tiered-storage/pom.xml b/tiered-storage/pom.xml
index f36d1b630ab..3bcb20a98a7 100644
--- a/tiered-storage/pom.xml
+++ b/tiered-storage/pom.xml
@@ -25,7 +25,7 @@
<parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
- <version>4.0.9</version>
+ <version>4.0.10-SNAPSHOT</version>
</parent>
<artifactId>tiered-storage-parent</artifactId>
(pulsar) 06/11: [fix][broker] Return failed future instead of throwing exception in async methods (#25289)
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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 0a717fb5765039a9e5456d682255b9edb1134021
Author: Hao Zhang <zh...@cmss.chinamobile.com>
AuthorDate: Tue Mar 10 01:04:10 2026 +0800
[fix][broker] Return failed future instead of throwing exception in async methods (#25289)
Co-authored-by: 张浩 <zh...@100.me>
(cherry picked from commit 35dae97c0bf5e3810d92ba12922e98a85d9634ec)
---
.../authorization/PulsarAuthorizationProvider.java | 2 +-
.../pulsar/broker/admin/impl/BrokersBase.java | 4 +--
.../pulsar/broker/admin/impl/NamespacesBase.java | 5 +--
.../broker/admin/impl/PersistentTopicsBase.java | 37 ++++++++++++----------
.../pulsar/broker/delayed/bucket/Bucket.java | 5 +--
.../channel/ServiceUnitStateChannelImpl.java | 2 +-
.../pulsar/broker/namespace/NamespaceService.java | 16 +++++++---
.../pulsar/broker/service/BrokerService.java | 8 +++--
.../SystemTopicBasedTopicPoliciesService.java | 4 ++-
.../org/apache/pulsar/broker/service/Topic.java | 4 ++-
.../service/nonpersistent/NonPersistentTopic.java | 3 +-
.../prometheus/PrometheusMetricsGenerator.java | 3 +-
.../pulsar/broker/systopic/SystemTopicClient.java | 3 +-
.../pulsar/broker/web/PulsarWebResource.java | 6 +++-
.../compaction/PulsarCompactionServiceFactory.java | 10 ++++--
.../compaction/StrategicTwoPhaseCompactor.java | 2 +-
.../client/impl/PulsarChannelInitializer.java | 12 ++++---
.../org/apache/pulsar/common/util/FutureUtil.java | 21 ++++++++----
.../pulsar/common/util/netty/ChannelFutures.java | 6 ++--
.../pulsar/common/util/netty/NettyFutureUtil.java | 10 ++++--
.../common/util/netty/ChannelFuturesTest.java | 12 +++++--
21 files changed, 116 insertions(+), 59 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
index 976e7b7ee12..e4f2ac7e8cd 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
@@ -326,7 +326,7 @@ public class PulsarAuthorizationProvider implements AuthorizationProvider {
private CompletableFuture<Void> checkNamespace(Stream<String> namespaces) {
boolean sameNamespace = namespaces.distinct().count() == 1;
if (!sameNamespace) {
- throw new IllegalArgumentException("The namespace should be the same");
+ return FutureUtil.failedFuture(new IllegalArgumentException("The namespace should be the same"));
}
return CompletableFuture.completedFuture(null);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
index 1e4e4ff66dd..3ee2a1285d3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java
@@ -420,7 +420,8 @@ public class BrokersBase extends AdminResource {
private CompletableFuture<Void> internalDeleteDynamicConfigurationOnMetadataAsync(String configName) {
if (!pulsar().getBrokerService().isDynamicConfiguration(configName)) {
- throw new RestException(Status.PRECONDITION_FAILED, "Can't delete non-dynamic configuration");
+ return FutureUtil.failedFuture(
+ new RestException(Status.PRECONDITION_FAILED, "Can't delete non-dynamic configuration"));
} else {
return dynamicConfigurationResources().setDynamicConfigurationAsync(old -> {
if (old != null) {
@@ -536,4 +537,3 @@ public class BrokersBase extends AdminResource {
return CompletableFuture.completedFuture(null);
}
}
-
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 b9ec275f79c..bb4cac85087 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
@@ -748,7 +748,8 @@ public abstract class NamespacesBase extends AdminResource {
private CompletableFuture<Void> checkNamespace(Stream<String> namespaces) {
boolean sameNamespace = namespaces.distinct().count() == 1;
if (!sameNamespace) {
- throw new RestException(Status.BAD_REQUEST, "The namespace should be the same");
+ return FutureUtil.failedFuture(
+ new RestException(Status.BAD_REQUEST, "The namespace should be the same"));
}
return CompletableFuture.completedFuture(null);
}
@@ -1971,7 +1972,7 @@ public abstract class NamespacesBase extends AdminResource {
+ " Repl clusters: %s, allowed clusters: %s",
ns.toString(), policies.replication_clusters, policies.allowed_clusters);
log.info(msg);
- throw new RestException(Status.BAD_REQUEST, msg);
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, msg));
}
pulsar().getBrokerService().setCurrentClusterAllowedIfNoClusterIsAllowed(ns, policies);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index 0a683f4b872..3e625a8d5ca 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -1070,8 +1070,8 @@ public class PersistentTopicsBase extends AdminResource {
protected CompletableFuture<Void> internalSetMaxUnackedMessagesOnSubscription(Integer maxUnackedNumToSet,
boolean isGlobal) {
if (maxUnackedNumToSet != null && maxUnackedNumToSet < 0) {
- throw new RestException(Status.PRECONDITION_FAILED,
- "maxUnackedNum must be 0 or more");
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "maxUnackedNum must be 0 or more"));
}
return pulsar().getTopicPoliciesService()
@@ -1095,8 +1095,8 @@ public class PersistentTopicsBase extends AdminResource {
protected CompletableFuture<Void> internalSetMaxUnackedMessagesOnConsumer(Integer maxUnackedNumToSet,
boolean isGlobal) {
if (maxUnackedNumToSet != null && maxUnackedNumToSet < 0) {
- throw new RestException(Status.PRECONDITION_FAILED,
- "maxUnackedNum must be 0 or more");
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "maxUnackedNum must be 0 or more"));
}
return pulsar().getTopicPoliciesService()
@@ -1108,7 +1108,8 @@ public class PersistentTopicsBase extends AdminResource {
protected CompletableFuture<Void> internalSetDeduplicationSnapshotInterval(Integer intervalToSet,
boolean isGlobal) {
if (intervalToSet != null && intervalToSet < 0) {
- throw new RestException(Status.PRECONDITION_FAILED, "interval must be 0 or more");
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "interval must be 0 or more"));
}
return pulsar().getTopicPoliciesService()
.updateTopicPoliciesAsync(topicName, isGlobal, intervalToSet == null, policies -> {
@@ -3616,9 +3617,9 @@ public class PersistentTopicsBase extends AdminResource {
protected CompletableFuture<Void> internalSetMaxMessageSize(Integer maxMessageSizeToSet, boolean isGlobal) {
if (maxMessageSizeToSet != null && (maxMessageSizeToSet < 0
|| maxMessageSizeToSet > config().getMaxMessageSize())) {
- throw new RestException(Status.PRECONDITION_FAILED
- , "topic-level maxMessageSize must be greater than or equal to 0 "
- + "and must be smaller than that in the broker-level");
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "topic-level maxMessageSize must be greater than or equal to 0 "
+ + "and must be smaller than that in the broker-level"));
}
return pulsar().getTopicPoliciesService()
@@ -3646,8 +3647,8 @@ public class PersistentTopicsBase extends AdminResource {
protected CompletableFuture<Void> internalSetMaxProducers(Integer maxProducersToSet, boolean isGlobal) {
if (maxProducersToSet != null && maxProducersToSet < 0) {
- throw new RestException(Status.PRECONDITION_FAILED,
- "maxProducers must be 0 or more");
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "maxProducers must be 0 or more"));
}
return pulsar().getTopicPoliciesService()
.updateTopicPoliciesAsync(topicName, isGlobal, maxProducersToSet == null, policies -> {
@@ -3663,8 +3664,8 @@ public class PersistentTopicsBase extends AdminResource {
protected CompletableFuture<Void> internalSetMaxSubscriptionsPerTopic(Integer maxSubscriptionsToSet,
boolean isGlobal) {
if (maxSubscriptionsToSet != null && maxSubscriptionsToSet < 0) {
- throw new RestException(Status.PRECONDITION_FAILED,
- "maxSubscriptionsPerTopic must be 0 or more");
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "maxSubscriptionsPerTopic must be 0 or more"));
}
return pulsar().getTopicPoliciesService()
@@ -3748,8 +3749,8 @@ public class PersistentTopicsBase extends AdminResource {
protected CompletableFuture<Void> internalSetMaxConsumers(Integer maxConsumersToSet, boolean isGlobal) {
if (maxConsumersToSet != null && maxConsumersToSet < 0) {
- throw new RestException(Status.PRECONDITION_FAILED,
- "maxConsumers must be 0 or more");
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "maxConsumers must be 0 or more"));
}
return pulsar().getTopicPoliciesService()
.updateTopicPoliciesAsync(topicName, isGlobal, maxConsumersToSet == null, policies -> {
@@ -4650,7 +4651,7 @@ public class PersistentTopicsBase extends AdminResource {
futures.add(pulsar().getAdminClient().topics().trimTopicAsync(topicNamePartition.toString()));
} catch (Exception e) {
log.error("[{}] Failed to trim topic {}", clientAppId(), topicNamePartition, e);
- throw new RestException(e);
+ return FutureUtil.failedFuture(new RestException(e));
}
}
return FutureUtil.waitForAll(futures).thenAccept(asyncResponse::resume);
@@ -4764,7 +4765,8 @@ public class PersistentTopicsBase extends AdminResource {
protected CompletableFuture<Void> internalSetMaxConsumersPerSubscription(
Integer maxConsumersPerSubscriptionToSet, boolean isGlobal) {
if (maxConsumersPerSubscriptionToSet != null && maxConsumersPerSubscriptionToSet < 0) {
- throw new RestException(Status.PRECONDITION_FAILED, "Invalid value for maxConsumersPerSubscription");
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid value for maxConsumersPerSubscription"));
}
return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, isGlobal, false, policies -> {
policies.setMaxConsumersPerSubscription(maxConsumersPerSubscriptionToSet);
@@ -4793,7 +4795,8 @@ public class PersistentTopicsBase extends AdminResource {
protected CompletableFuture<Void> internalSetCompactionThreshold(Long compactionThresholdToSet, boolean isGlobal) {
if (compactionThresholdToSet != null && compactionThresholdToSet < 0) {
- throw new RestException(Status.PRECONDITION_FAILED, "Invalid value for compactionThreshold");
+ return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ "Invalid value for compactionThreshold"));
}
return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, isGlobal, false, policies -> {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/Bucket.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/Bucket.java
index a1693b1553d..776f99b120c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/Bucket.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/Bucket.java
@@ -23,7 +23,6 @@ import static org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTrack
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import lombok.AllArgsConstructor;
@@ -158,7 +157,9 @@ abstract class Bucket {
}
private CompletableFuture<Void> putBucketKeyId(String bucketKey, Long bucketId) {
- Objects.requireNonNull(bucketId);
+ if (bucketId == null) {
+ return FutureUtil.failedFuture(new NullPointerException("Expected bucketId should not be null"));
+ }
return sequencer.sequential(() -> {
return executeWithRetry(() -> cursor.putCursorProperty(bucketKey, String.valueOf(bucketId)),
ManagedLedgerException.BadVersionException.class, MaxRetryTimes);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
index 8b48e4b4516..a0e9f1f73e2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
@@ -671,7 +671,7 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
private CompletableFuture<Void> publishOverrideEventAsync(String serviceUnit,
ServiceUnitStateData override) {
if (!validateChannelState(Started, true)) {
- throw new IllegalStateException("Invalid channel state:" + channelState.name());
+ return FutureUtil.failedFuture(new IllegalStateException("Invalid channel state:" + channelState.name()));
}
EventType eventType = EventType.Override;
eventCounters.get(eventType).getTotal().incrementAndGet();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index b5dbddc2ea3..f17526416e5 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -1143,8 +1143,12 @@ public class NamespaceService implements AutoCloseable {
*/
public CompletableFuture<Void> updateNamespaceBundlesForPolicies(NamespaceName nsname,
NamespaceBundles nsBundles) {
- Objects.requireNonNull(nsname);
- Objects.requireNonNull(nsBundles);
+ if (nsname == null) {
+ return FutureUtil.failedFuture(new NullPointerException("Expected NamespaceName should not be null"));
+ }
+ if (nsBundles == null) {
+ return FutureUtil.failedFuture(new NullPointerException("Expected NamespaceBundles should not be null"));
+ }
return pulsar.getPulsarResources().getNamespaceResources().getPoliciesAsync(nsname).thenCompose(policies -> {
if (policies.isPresent()) {
@@ -1170,8 +1174,12 @@ public class NamespaceService implements AutoCloseable {
* @param nsBundles the new namespace bundles
*/
public CompletableFuture<Void> updateNamespaceBundles(NamespaceName nsname, NamespaceBundles nsBundles) {
- Objects.requireNonNull(nsname);
- Objects.requireNonNull(nsBundles);
+ if (nsname == null) {
+ return FutureUtil.failedFuture(new NullPointerException("Expected NamespaceName should not be null"));
+ }
+ if (nsBundles == null) {
+ return FutureUtil.failedFuture(new NullPointerException("Expected NamespaceBundles should not be null"));
+ }
LocalPolicies localPolicies = nsBundles.toLocalPolicies();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 7e9819f7f04..27008e96b37 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -1983,7 +1983,9 @@ public class BrokerService implements Closeable {
}
public CompletableFuture<ManagedLedgerConfig> getManagedLedgerConfig(@NonNull TopicName topicName) {
- requireNonNull(topicName);
+ if (topicName == null) {
+ return FutureUtil.failedFuture(new NullPointerException("topicName"));
+ }
NamespaceName namespace = topicName.getNamespaceObject();
ServiceConfiguration serviceConfig = pulsar.getConfiguration();
@@ -3783,7 +3785,9 @@ public class BrokerService implements Closeable {
}
public @NonNull CompletableFuture<Boolean> isAllowAutoSubscriptionCreationAsync(@NonNull TopicName tpName) {
- requireNonNull(tpName);
+ if (tpName == null) {
+ return FutureUtil.failedFuture(new NullPointerException("tpName"));
+ }
// Policies priority: topic level -> namespace level -> broker level
if (ExtensibleLoadManagerImpl.isInternalTopic(tpName.toString())) {
return CompletableFuture.completedFuture(true);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
index c3d88b9c723..ad18af308e2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
@@ -565,7 +565,9 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
@VisibleForTesting
@NonNull CompletableFuture<Boolean> prepareInitPoliciesCacheAsync(@NonNull NamespaceName namespace) {
- requireNonNull(namespace);
+ if (namespace == null) {
+ return FutureUtil.failedFuture(new NullPointerException("Expected NamespaceName should not be null"));
+ }
if (closed.get()) {
return CompletableFuture.completedFuture(false);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
index 8f66d9c0e3e..fd3ffb1a34c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
@@ -44,6 +44,7 @@ import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.stats.TopicStatsImpl;
import org.apache.pulsar.common.protocol.schema.SchemaData;
import org.apache.pulsar.common.protocol.schema.SchemaVersion;
+import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
import org.apache.pulsar.utils.StatsOutputStream;
@@ -291,7 +292,8 @@ public interface Topic {
* Get the last message position that can be dispatch.
*/
default CompletableFuture<Position> getLastDispatchablePosition() {
- throw new UnsupportedOperationException("getLastDispatchablePosition is not supported by default");
+ return FutureUtil.failedFuture(
+ new UnsupportedOperationException("getLastDispatchablePosition is not supported by default"));
}
CompletableFuture<MessageId> getLastMessageId();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
index 1931a09497e..e79c642e7b7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
@@ -1219,7 +1219,8 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol
@Override
public CompletableFuture<MessageId> getLastMessageId() {
- throw new UnsupportedOperationException("getLastMessageId is not supported on non-persistent topic");
+ return FutureUtil.failedFuture(
+ new UnsupportedOperationException("getLastMessageId is not supported on non-persistent topic"));
}
private static final Logger log = LoggerFactory.getLogger(NonPersistentTopic.class);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
index 97d5a7bc953..2bb6372c1cb 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
@@ -60,6 +60,7 @@ import org.apache.pulsar.broker.storage.BookkeeperManagedLedgerStorageClass;
import org.apache.pulsar.broker.storage.ManagedLedgerStorageClass;
import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
import org.apache.pulsar.common.stats.Metrics;
+import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.SimpleTextOutputStream;
/**
@@ -139,7 +140,7 @@ public class PrometheusMetricsGenerator implements AutoCloseable {
public synchronized CompletableFuture<ByteBuf> getCompressedBuffer(Executor executor) {
if (released) {
- throw new IllegalStateException("Already released!");
+ return FutureUtil.failedFuture(new IllegalStateException("Already released!"));
}
if (compressedBuffer == null) {
compressedBuffer = new CompletableFuture<>();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/SystemTopicClient.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/SystemTopicClient.java
index 88ca099b4ca..20521780bff 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/SystemTopicClient.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/systopic/SystemTopicClient.java
@@ -25,6 +25,7 @@ import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.util.FutureUtil;
/**
* Pulsar system topic.
@@ -123,7 +124,7 @@ public interface SystemTopicClient<T> {
* @return message id future
*/
default CompletableFuture<MessageId> deleteAsync(String key, T t) {
- throw new UnsupportedOperationException("Unsupported operation");
+ return FutureUtil.failedFuture(new UnsupportedOperationException("Unsupported operation"));
}
/**
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
index 65489eaa34b..9291097d271 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
@@ -213,7 +213,11 @@ public abstract class PulsarWebResource {
isClientAuthenticated(appId), appId);
}
String originalPrincipal = originalPrincipal();
- validateOriginalPrincipal(appId, originalPrincipal);
+ try {
+ validateOriginalPrincipal(appId, originalPrincipal);
+ } catch (RestException e) {
+ return FutureUtil.failedFuture(e);
+ }
if (pulsar.getConfiguration().getProxyRoles().contains(appId)) {
BrokerService brokerService = pulsar.getBrokerService();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarCompactionServiceFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarCompactionServiceFactory.java
index ff0fa0bcd12..b44480378c6 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarCompactionServiceFactory.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarCompactionServiceFactory.java
@@ -19,13 +19,13 @@
package org.apache.pulsar.compaction;
import com.google.common.annotations.VisibleForTesting;
-import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import lombok.AccessLevel;
import lombok.Getter;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.common.util.FutureUtil;
import org.jspecify.annotations.NonNull;
import org.jspecify.annotations.Nullable;
@@ -61,14 +61,18 @@ public class PulsarCompactionServiceFactory implements CompactionServiceFactory
@Override
public CompletableFuture<Void> initialize(@NonNull PulsarService pulsarService) {
- Objects.requireNonNull(pulsarService);
+ if (pulsarService == null) {
+ return FutureUtil.failedFuture(new NullPointerException("Expected pulsarService should not be null"));
+ }
this.pulsarService = pulsarService;
return CompletableFuture.completedFuture(null);
}
@Override
public CompletableFuture<TopicCompactionService> newTopicCompactionService(@NonNull String topic) {
- Objects.requireNonNull(topic);
+ if (topic == null) {
+ return FutureUtil.failedFuture(new NullPointerException("Expected topic should not be null"));
+ }
PulsarTopicCompactionService pulsarTopicCompactionService =
new PulsarTopicCompactionService(topic, pulsarService.getBookKeeperClient(), () -> {
try {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java
index 1b54092d9aa..3c674975742 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/StrategicTwoPhaseCompactor.java
@@ -76,7 +76,7 @@ public class StrategicTwoPhaseCompactor extends PublishingOrderCompactor {
}
public CompletableFuture<Long> compact(String topic) {
- throw new UnsupportedOperationException();
+ return FutureUtil.failedFuture(new UnsupportedOperationException());
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java
index 0d3f8a4c619..84198967601 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java
@@ -27,7 +27,6 @@ import io.netty.handler.proxy.Socks5ProxyHandler;
import io.netty.handler.ssl.SslHandler;
import java.net.InetSocketAddress;
import java.util.Map;
-import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
@@ -40,6 +39,7 @@ import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.apache.pulsar.common.protocol.ByteBufPair;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.protocol.FrameDecoderUtil;
+import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.PulsarSslConfiguration;
import org.apache.pulsar.common.util.PulsarSslFactory;
import org.apache.pulsar.common.util.SecurityUtility;
@@ -106,10 +106,14 @@ public class PulsarChannelInitializer extends ChannelInitializer<SocketChannel>
* @return a {@link CompletableFuture} that completes when the TLS is set up.
*/
CompletableFuture<Channel> initTls(Channel ch, InetSocketAddress sniHost) {
- Objects.requireNonNull(ch, "A channel is required");
- Objects.requireNonNull(sniHost, "A sniHost is required");
+ if (ch == null) {
+ return FutureUtil.failedFuture(new NullPointerException("A channel is required"));
+ }
+ if (sniHost == null) {
+ return FutureUtil.failedFuture(new NullPointerException("A sniHost is required"));
+ }
if (!tlsEnabled) {
- throw new IllegalStateException("TLS is not enabled in client configuration");
+ return FutureUtil.failedFuture(new IllegalStateException("TLS is not enabled in client configuration"));
}
CompletableFuture<Channel> initTlsFuture = new CompletableFuture<>();
ch.eventLoop().execute(() -> {
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
index ae784a1b18d..1b43d17d9d1 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java
@@ -24,7 +24,6 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
-import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
@@ -228,10 +227,13 @@ public class FutureUtil {
}
/**
- * @throws NullPointerException NPE when param is null
+ * @return a {@link CompletableFuture} representing the newly scheduled task,
+ * or one completed exceptionally with {@link NullPointerException} if param is null.
*/
public synchronized CompletableFuture<T> sequential(Supplier<CompletableFuture<T>> newTask) {
- Objects.requireNonNull(newTask);
+ if (newTask == null) {
+ return failedFuture(new NullPointerException("Expected Supplier should not be null"));
+ }
if (sequencerFuture.isDone()) {
if (sequencerFuture.isCompletedExceptionally() && allowExceptionBreakChain) {
return sequencerFuture;
@@ -282,13 +284,18 @@ public class FutureUtil {
}
/**
- * @throws RejectedExecutionException if this task cannot be accepted for execution
- * @throws NullPointerException if one of params is null
+ * @return a {@link CompletableFuture} representing the asynchronous composition.
+ * The returned future is completed exceptionally with {@link NullPointerException} if one of params is null,
+ * or with {@link RejectedExecutionException} if the task cannot be accepted for execution.
*/
public static <T> @NonNull CompletableFuture<T> composeAsync(Supplier<CompletableFuture<T>> futureSupplier,
Executor executor) {
- Objects.requireNonNull(futureSupplier);
- Objects.requireNonNull(executor);
+ if (futureSupplier == null) {
+ return failedFuture(new NullPointerException("Expected Supplier should not be null"));
+ }
+ if (executor == null) {
+ return failedFuture(new NullPointerException("Expected Executor should not be null"));
+ }
final CompletableFuture<T> future = new CompletableFuture<>();
try {
executor.execute(() -> futureSupplier.get().whenComplete((result, error) -> {
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/ChannelFutures.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/ChannelFutures.java
index 294c0db99ea..481d47a21c2 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/ChannelFutures.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/ChannelFutures.java
@@ -20,8 +20,8 @@ package org.apache.pulsar.common.util.netty;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
-import java.util.Objects;
import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.common.util.FutureUtil;
/**
* Static utility methods for operating on {@link ChannelFuture}s.
@@ -41,7 +41,9 @@ public class ChannelFutures {
* and completes exceptionally if the channelFuture completes with a {@link Throwable}
*/
public static CompletableFuture<Channel> toCompletableFuture(ChannelFuture channelFuture) {
- Objects.requireNonNull(channelFuture, "channelFuture cannot be null");
+ if (channelFuture == null) {
+ return FutureUtil.failedFuture(new NullPointerException("channelFuture cannot be null"));
+ }
CompletableFuture<Channel> adapter = new CompletableFuture<>();
if (channelFuture.isDone()) {
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/NettyFutureUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/NettyFutureUtil.java
index 8686a381679..109fa2fa265 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/NettyFutureUtil.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/NettyFutureUtil.java
@@ -19,8 +19,8 @@
package org.apache.pulsar.common.util.netty;
import io.netty.util.concurrent.Future;
-import java.util.Objects;
import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.common.util.FutureUtil;
/**
* Contains utility methods for working with Netty Futures.
@@ -34,7 +34,9 @@ public class NettyFutureUtil {
* @return converted future instance
*/
public static <V> CompletableFuture<V> toCompletableFuture(Future<V> future) {
- Objects.requireNonNull(future, "future cannot be null");
+ if (future == null) {
+ return FutureUtil.failedFuture(new NullPointerException("future cannot be null"));
+ }
CompletableFuture<V> adapter = new CompletableFuture<>();
if (future.isDone()) {
@@ -62,7 +64,9 @@ public class NettyFutureUtil {
* @return converted future instance
*/
public static CompletableFuture<Void> toCompletableFutureVoid(Future<?> future) {
- Objects.requireNonNull(future, "future cannot be null");
+ if (future == null) {
+ return FutureUtil.failedFuture(new NullPointerException("future cannot be null"));
+ }
CompletableFuture<Void> adapter = new CompletableFuture<>();
if (future.isDone()) {
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/ChannelFuturesTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/ChannelFuturesTest.java
index 31c0eb4af2b..65b9e7b6339 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/ChannelFuturesTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/ChannelFuturesTest.java
@@ -23,6 +23,7 @@ import io.netty.channel.Channel;
import io.netty.channel.DefaultChannelPromise;
import io.netty.channel.DefaultEventLoop;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import org.mockito.Mock;
@@ -65,9 +66,16 @@ public class ChannelFuturesTest {
channelFuture = new DefaultChannelPromise(channel);
}
- @Test(expectedExceptions = NullPointerException.class)
+ @Test
public void toCompletableFuture_shouldRequireNonNullArgument() {
- ChannelFutures.toCompletableFuture(null);
+ CompletableFuture<Channel> future = ChannelFutures.toCompletableFuture(null);
+ Assert.assertTrue(future.isCompletedExceptionally());
+ try {
+ future.join();
+ Assert.fail("Expected NullPointerException");
+ } catch (CompletionException e) {
+ Assert.assertTrue(e.getCause() instanceof NullPointerException);
+ }
}
@Test
(pulsar) 04/11: [fix][admin] Refactor namespace anti affinity group sync operations to async in rest api (#25086)
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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 125d1d194f4084c7eaa4a32e073096da5ac6e1bd
Author: Oneby Wang <44...@users.noreply.github.com>
AuthorDate: Wed Dec 17 19:16:20 2025 +0800
[fix][admin] Refactor namespace anti affinity group sync operations to async in rest api (#25086)
Co-authored-by: oneby-wang <on...@qq.com>
(cherry picked from commit e041fab73c1b9a6bb93457d186fc0a937beafd53)
---
.../broker/resources/LocalPoliciesResources.java | 6 +-
.../apache/pulsar/broker/admin/AdminResource.java | 17 +-
.../pulsar/broker/admin/impl/NamespacesBase.java | 186 +++++++--------------
.../apache/pulsar/broker/admin/v1/Namespaces.java | 55 ++++--
.../apache/pulsar/broker/admin/v2/Namespaces.java | 58 +++++--
.../apache/pulsar/broker/admin/NamespacesTest.java | 80 +++++++++
.../pulsar/broker/admin/NamespacesV2Test.java | 95 ++++++++++-
7 files changed, 332 insertions(+), 165 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/LocalPoliciesResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/LocalPoliciesResources.java
index b7ef19ccbe8..8e7b0ab0b1e 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/LocalPoliciesResources.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/LocalPoliciesResources.java
@@ -37,9 +37,9 @@ public class LocalPoliciesResources extends BaseResources<LocalPolicies> {
super(localStore, LocalPolicies.class, operationTimeoutSec);
}
- public void setLocalPolicies(NamespaceName ns, Function<LocalPolicies, LocalPolicies> modifyFunction)
- throws MetadataStoreException {
- set(joinPath(LOCAL_POLICIES_ROOT, ns.toString()), modifyFunction);
+ public CompletableFuture<Void> setLocalPoliciesAsync(NamespaceName ns,
+ Function<LocalPolicies, LocalPolicies> modifyFunction) {
+ return setAsync(joinPath(LOCAL_POLICIES_ROOT, ns.toString()), modifyFunction);
}
public Optional<LocalPolicies> getLocalPolicies(NamespaceName ns) throws MetadataStoreException{
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
index d35ea51d283..103181a7c73 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.admin;
+import static org.apache.commons.lang3.StringUtils.isBlank;
import com.fasterxml.jackson.databind.ObjectReader;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.google.errorprone.annotations.CanIgnoreReturnValue;
@@ -497,14 +498,12 @@ public abstract class AdminResource extends PulsarWebResource {
});
}
- protected void validateClusterExists(String cluster) {
- try {
- if (!clusterResources().getCluster(cluster).isPresent()) {
+ protected CompletableFuture<Void> validateClusterExistsAsync(String cluster) {
+ return clusterResources().clusterExistsAsync(cluster).thenAccept(clusterExist -> {
+ if (!clusterExist) {
throw new RestException(Status.PRECONDITION_FAILED, "Cluster " + cluster + " does not exist.");
}
- } catch (Exception e) {
- throw new RestException(e);
- }
+ });
}
protected Policies getNamespacePolicies(String tenant, String cluster, String namespace) {
@@ -853,6 +852,12 @@ public abstract class AdminResource extends PulsarWebResource {
}
}
+ protected void checkNotBlank(String str, String errorMessage) {
+ if (isBlank(str)) {
+ throw new RestException(Status.PRECONDITION_FAILED, errorMessage);
+ }
+ }
+
protected boolean isManagedLedgerNotFoundException(Throwable cause) {
return cause instanceof ManagedLedgerException.MetadataNotFoundException
|| cause instanceof MetadataStoreException.NotFoundException;
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 f28b07d8a9a..b9ec275f79c 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
@@ -18,7 +18,6 @@
*/
package org.apache.pulsar.broker.admin.impl;
-import static org.apache.commons.lang3.StringUtils.isBlank;
import static org.apache.pulsar.common.policies.data.PoliciesUtil.getBundles;
import com.google.common.collect.Sets;
import java.lang.reflect.Field;
@@ -1038,35 +1037,6 @@ public abstract class NamespacesBase extends AdminResource {
return internalSetBookieAffinityGroupAsync(null);
}
- @Deprecated
- protected BookieAffinityGroupData internalGetBookieAffinityGroup() {
- validateSuperUserAccess();
-
- 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());
- }
- try {
- final BookieAffinityGroupData bookkeeperAffinityGroup = getLocalPolicies().getLocalPolicies(namespaceName)
- .orElseThrow(() -> new RestException(Status.NOT_FOUND,
- "Namespace local-policies does not exist")).bookieAffinityGroup;
- return bookkeeperAffinityGroup;
- } catch (NotFoundException e) {
- log.warn("[{}] Failed to get local-policy configuration for namespace {}: does not exist",
- clientAppId(), namespaceName);
- throw new RestException(Status.NOT_FOUND, "Namespace policies does not exist");
- } catch (RestException re) {
- throw re;
- } catch (Exception e) {
- log.error("[{}] Failed to get local-policy configuration for namespace {}", clientAppId(),
- namespaceName, e);
- throw new RestException(e);
- }
- }
-
protected CompletableFuture<BookieAffinityGroupData> internalGetBookieAffinityGroupAsync() {
return validateSuperUserAccessAsync().thenCompose(__ -> {
if (namespaceName.isGlobal()) {
@@ -1077,9 +1047,8 @@ public abstract class NamespacesBase extends AdminResource {
unused -> validateClusterForTenantAsync(namespaceName.getTenant(), namespaceName.getCluster()));
}
}).thenCompose(__ -> getLocalPolicies().getLocalPoliciesAsync(namespaceName))
- .thenApply(policies -> policies.orElseThrow(
- () -> new RestException(Status.NOT_FOUND, "Namespace local-policies does not exist"))
- .bookieAffinityGroup);
+ .thenApply(policies -> policies.orElseThrow(() -> new RestException(Status.NOT_FOUND,
+ "Namespace local-policies does not exist")).bookieAffinityGroup);
}
private CompletableFuture<Void> validateLeaderBrokerAsync() {
@@ -1823,104 +1792,69 @@ public abstract class NamespacesBase extends AdminResource {
internalSetPolicies("delayed_delivery_policies", delayedDeliveryPolicies);
}
- protected void internalSetNamespaceAntiAffinityGroup(String antiAffinityGroup) {
- validateNamespacePolicyOperation(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE);
- checkNotNull(antiAffinityGroup, "AntiAffinityGroup should not be null");
- validatePoliciesReadOnlyAccess();
-
- log.info("[{}] Setting anti-affinity group {} for {}", clientAppId(), antiAffinityGroup, namespaceName);
-
- if (isBlank(antiAffinityGroup)) {
- throw new RestException(Status.PRECONDITION_FAILED, "antiAffinityGroup can't be empty");
- }
-
- try {
- getLocalPolicies().setLocalPoliciesWithCreate(namespaceName, (lp)->
- lp.map(policies -> new LocalPolicies(policies.bundles,
- policies.bookieAffinityGroup,
- antiAffinityGroup,
- policies.migrated))
- .orElseGet(() -> new LocalPolicies(getDefaultBundleData(), null, antiAffinityGroup))
- );
- log.info("[{}] Successfully updated local-policies configuration: namespace={}, map={}", clientAppId(),
- namespaceName, antiAffinityGroup);
- } catch (RestException re) {
- throw re;
- } catch (Exception e) {
- log.error("[{}] Failed to update local-policy configuration for namespace {}", clientAppId(), namespaceName,
- e);
- throw new RestException(e);
- }
- }
-
- protected String internalGetNamespaceAntiAffinityGroup() {
- validateNamespacePolicyOperation(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ);
-
- try {
- return getLocalPolicies()
- .getLocalPolicies(namespaceName)
- .orElseGet(() -> new LocalPolicies(getBundles(config().getDefaultNumberOfNamespaceBundles())
- , null, null)).namespaceAntiAffinityGroup;
- } catch (Exception e) {
- log.error("[{}] Failed to get the antiAffinityGroup of namespace {}", clientAppId(), namespaceName, e);
- throw new RestException(Status.NOT_FOUND, "Couldn't find namespace policies");
- }
- }
-
- protected void internalRemoveNamespaceAntiAffinityGroup() {
- validateNamespacePolicyOperation(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE);
- validatePoliciesReadOnlyAccess();
-
- log.info("[{}] Deleting anti-affinity group for {}", clientAppId(), namespaceName);
-
- try {
- getLocalPolicies().setLocalPolicies(namespaceName, (policies)->
- new LocalPolicies(policies.bundles,
- policies.bookieAffinityGroup,
- null,
- policies.migrated));
- log.info("[{}] Successfully removed anti-affinity group for a namespace={}", clientAppId(), namespaceName);
- } catch (Exception e) {
- log.error("[{}] Failed to remove anti-affinity group for namespace {}", clientAppId(), namespaceName, e);
- throw new RestException(e);
- }
+ protected CompletableFuture<Void> internalSetNamespaceAntiAffinityGroupAsync(String antiAffinityGroup) {
+ checkNotNull(antiAffinityGroup, "Anti-affinity group should not be null");
+ checkNotBlank(antiAffinityGroup, "Anti-affinity group can't be empty");
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE)
+ .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync()).thenCompose(
+ __ -> getDefaultBundleDataAsync().thenCompose(
+ defaultBundleData -> getLocalPolicies().setLocalPoliciesWithCreateAsync(namespaceName,
+ oldPolicies -> oldPolicies.map(policies -> new LocalPolicies(policies.bundles,
+ policies.bookieAffinityGroup, antiAffinityGroup,
+ policies.migrated))
+ .orElseGet(() -> new LocalPolicies(defaultBundleData, null,
+ antiAffinityGroup)))))
+ .thenAccept(__ -> log.info(
+ "[{}] Successfully updated namespace anti-affinity group, namespace={}, anti-affinity"
+ + " group={}", clientAppId(), namespaceName, antiAffinityGroup));
+ }
+
+ protected CompletableFuture<String> internalGetNamespaceAntiAffinityGroupAsync() {
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ)
+ .thenCompose(__ -> getLocalPolicies().getLocalPoliciesAsync(namespaceName)
+ .thenApply(policiesOpt -> policiesOpt.map(localPolicies -> localPolicies.namespaceAntiAffinityGroup)
+ .orElse(null)));
+ }
+
+ protected CompletableFuture<Void> internalRemoveNamespaceAntiAffinityGroupAsync() {
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE)
+ .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync())
+ .thenCompose(__ -> {
+ log.info("[{}] Removing anti-affinity group for namespace: {}", clientAppId(), namespaceName);
+ return getLocalPolicies().setLocalPoliciesAsync(namespaceName,
+ (policies) -> new LocalPolicies(policies.bundles, policies.bookieAffinityGroup, null,
+ policies.migrated));
+ })
+ .thenAccept(__ -> log.info("[{}] Successfully removed anti-affinity group for namespace: {}",
+ clientAppId(), namespaceName));
}
- protected List<String> internalGetAntiAffinityNamespaces(String cluster, String antiAffinityGroup,
- String tenant) {
- validateNamespacePolicyOperation(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ);
+ protected CompletableFuture<List<String>> internalGetAntiAffinityNamespacesAsync(String cluster,
+ String antiAffinityGroup,
+ String tenant) {
checkNotNull(cluster, "Cluster should not be null");
- checkNotNull(antiAffinityGroup, "AntiAffinityGroup should not be null");
+ checkNotNull(antiAffinityGroup, "Anti-affinity group should not be null");
checkNotNull(tenant, "Tenant should not be null");
+ checkNotBlank(antiAffinityGroup, "Anti-affinity group can't be empty");
- log.info("[{}]-{} Finding namespaces for {} in {}", clientAppId(), tenant, antiAffinityGroup, cluster);
-
- if (isBlank(antiAffinityGroup)) {
- throw new RestException(Status.PRECONDITION_FAILED, "anti-affinity group can't be empty.");
- }
- validateClusterExists(cluster);
-
- try {
- List<String> namespaces = tenantResources().getListOfNamespaces(tenant);
-
- return namespaces.stream().filter(ns -> {
- Optional<LocalPolicies> policies;
- try {
- policies = getLocalPolicies().getLocalPolicies(NamespaceName.get(ns));
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
-
- String storedAntiAffinityGroup = policies.orElseGet(() ->
- new LocalPolicies(getBundles(config().getDefaultNumberOfNamespaceBundles()),
- null, null)).namespaceAntiAffinityGroup;
- return antiAffinityGroup.equalsIgnoreCase(storedAntiAffinityGroup);
- }).collect(Collectors.toList());
-
- } catch (Exception e) {
- log.warn("Failed to list of properties/namespace from global-zk", e);
- throw new RestException(e);
- }
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ)
+ .thenCompose(__ -> validateClusterExistsAsync(cluster))
+ .thenCompose(__ -> {
+ log.info("[{}]-{} Finding namespaces for {} in {}", clientAppId(), tenant, antiAffinityGroup,
+ cluster);
+ return tenantResources().getListOfNamespacesAsync(tenant).thenCompose(namespaces -> {
+ List<CompletableFuture<String>> nsFutures = namespaces.stream()
+ .map(ns -> getLocalPolicies().getLocalPoliciesAsync(NamespaceName.get(ns))
+ .thenApply(policiesOpt -> policiesOpt.map(
+ localPolicies -> localPolicies.namespaceAntiAffinityGroup).orElse(null))
+ .thenApply(antiAffinityGroup::equalsIgnoreCase)
+ .thenApply(equals -> equals ? ns : null)).toList();
+ CompletableFuture<Void> allFuture = FutureUtil.waitForAll(nsFutures);
+ return allFuture.thenApply(
+ unused -> nsFutures.stream().map(CompletableFuture::join).filter(Objects::nonNull)
+ .toList());
+ });
+ });
}
private boolean checkQuotas(Policies policies, RetentionPolicies retention) {
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 276ea457d2c..4ba8d6aa1da 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
@@ -583,10 +583,19 @@ public class Namespaces extends NamespacesBase {
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Property or cluster or namespace doesn't exist"),
@ApiResponse(code = 412, message = "Invalid antiAffinityGroup") })
- public void setNamespaceAntiAffinityGroup(@PathParam("property") String property,
- @PathParam("cluster") String cluster, @PathParam("namespace") String namespace, String antiAffinityGroup) {
+ public void setNamespaceAntiAffinityGroup(@Suspended AsyncResponse asyncResponse,
+ @PathParam("property") String property,
+ @PathParam("cluster") String cluster,
+ @PathParam("namespace") String namespace, String antiAffinityGroup) {
validateNamespaceName(property, cluster, namespace);
- internalSetNamespaceAntiAffinityGroup(antiAffinityGroup);
+ internalSetNamespaceAntiAffinityGroupAsync(antiAffinityGroup)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to set namespace anti-affinity group, tenant: {}, namespace: {}, "
+ + "antiAffinityGroup: {}", clientAppId(), property, namespace, antiAffinityGroup, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@GET
@@ -594,10 +603,19 @@ public class Namespaces extends NamespacesBase {
@ApiOperation(value = "Get anti-affinity group of a namespace.")
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Property or cluster or namespace doesn't exist") })
- public String getNamespaceAntiAffinityGroup(@PathParam("property") String property,
- @PathParam("cluster") String cluster, @PathParam("namespace") String namespace) {
+ public void getNamespaceAntiAffinityGroup(@Suspended AsyncResponse asyncResponse,
+ @PathParam("property") String property,
+ @PathParam("cluster") String cluster,
+ @PathParam("namespace") String namespace) {
validateNamespaceName(property, cluster, namespace);
- return internalGetNamespaceAntiAffinityGroup();
+ internalGetNamespaceAntiAffinityGroupAsync()
+ .thenAccept(asyncResponse::resume)
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to get namespace anti-affinity group, tenant: {}, namespace: {}",
+ clientAppId(), property, namespace, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@GET
@@ -606,10 +624,19 @@ public class Namespaces extends NamespacesBase {
+ " api can be only accessed by admin of any of the existing property")
@ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 412, message = "Cluster not exist/Anti-affinity group can't be empty.")})
- public List<String> getAntiAffinityNamespaces(@PathParam("cluster") String cluster,
+ public void getAntiAffinityNamespaces(@Suspended AsyncResponse asyncResponse,
+ @PathParam("cluster") String cluster,
@PathParam("group") String antiAffinityGroup,
@QueryParam("property") String property) {
- return internalGetAntiAffinityNamespaces(cluster, antiAffinityGroup, property);
+ internalGetAntiAffinityNamespacesAsync(cluster, antiAffinityGroup, property)
+ .thenAccept(asyncResponse::resume)
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to get all namespaces in cluster of given anti-affinity group, cluster: {}, "
+ + "tenant: {}, antiAffinityGroup: {}", clientAppId(), cluster, property, antiAffinityGroup,
+ ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@DELETE
@@ -618,11 +645,19 @@ public class Namespaces extends NamespacesBase {
@ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Namespace does not exist"),
@ApiResponse(code = 409, message = "Concurrent modification")})
- public void removeNamespaceAntiAffinityGroup(@PathParam("property") String property,
+ public void removeNamespaceAntiAffinityGroup(@Suspended AsyncResponse asyncResponse,
+ @PathParam("property") String property,
@PathParam("cluster") String cluster,
@PathParam("namespace") String namespace) {
validateNamespaceName(property, cluster, namespace);
- internalRemoveNamespaceAntiAffinityGroup();
+ internalRemoveNamespaceAntiAffinityGroupAsync()
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to remove namespace anti-affinity group, tenant: {}, namespace: {}",
+ clientAppId(), property, namespace, 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 5058eccc401..732cb315604 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
@@ -2144,13 +2144,21 @@ public class Namespaces extends NamespacesBase {
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Tenant or cluster or namespace doesn't exist"),
@ApiResponse(code = 412, message = "Invalid antiAffinityGroup")})
- public void setNamespaceAntiAffinityGroup(@PathParam("tenant") String tenant,
+ public void setNamespaceAntiAffinityGroup(@Suspended AsyncResponse asyncResponse,
+ @PathParam("tenant") String tenant,
@PathParam("namespace") String namespace,
@ApiParam(value = "Anti-affinity group for the specified namespace",
required = true)
- String antiAffinityGroup) {
+ String antiAffinityGroup) {
validateNamespaceName(tenant, namespace);
- internalSetNamespaceAntiAffinityGroup(antiAffinityGroup);
+ internalSetNamespaceAntiAffinityGroupAsync(antiAffinityGroup)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to set namespace anti-affinity group, tenant: {}, namespace: {}, "
+ + "antiAffinityGroup: {}", clientAppId(), tenant, namespace, antiAffinityGroup, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@GET
@@ -2158,10 +2166,18 @@ public class Namespaces extends NamespacesBase {
@ApiOperation(value = "Get anti-affinity group of a namespace.", response = String.class)
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Tenant or cluster or namespace doesn't exist") })
- public String getNamespaceAntiAffinityGroup(@PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace) {
+ public void getNamespaceAntiAffinityGroup(@Suspended AsyncResponse asyncResponse,
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace) {
validateNamespaceName(tenant, namespace);
- return internalGetNamespaceAntiAffinityGroup();
+ internalGetNamespaceAntiAffinityGroupAsync()
+ .thenAccept(asyncResponse::resume)
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to get namespace anti-affinity group, tenant: {}, namespace: {}",
+ clientAppId(), tenant, namespace, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@DELETE
@@ -2172,10 +2188,18 @@ public class Namespaces extends NamespacesBase {
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Namespace does not exist"),
@ApiResponse(code = 409, message = "Concurrent modification") })
- public void removeNamespaceAntiAffinityGroup(@PathParam("tenant") String tenant,
- @PathParam("namespace") String namespace) {
+ public void removeNamespaceAntiAffinityGroup(@Suspended AsyncResponse asyncResponse,
+ @PathParam("tenant") String tenant,
+ @PathParam("namespace") String namespace) {
validateNamespaceName(tenant, namespace);
- internalRemoveNamespaceAntiAffinityGroup();
+ internalRemoveNamespaceAntiAffinityGroupAsync()
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to remove namespace anti-affinity group, tenant: {}, namespace: {}",
+ clientAppId(), tenant, namespace, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@GET
@@ -2185,9 +2209,19 @@ public class Namespaces extends NamespacesBase {
response = String.class, responseContainer = "List")
@ApiResponses(value = {@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 412, message = "Cluster not exist/Anti-affinity group can't be empty.")})
- public List<String> getAntiAffinityNamespaces(@PathParam("cluster") String cluster,
- @PathParam("group") String antiAffinityGroup, @QueryParam("tenant") String tenant) {
- return internalGetAntiAffinityNamespaces(cluster, antiAffinityGroup, tenant);
+ public void getAntiAffinityNamespaces(@Suspended AsyncResponse asyncResponse,
+ @PathParam("cluster") String cluster,
+ @PathParam("group") String antiAffinityGroup,
+ @QueryParam("tenant") String tenant) {
+ internalGetAntiAffinityNamespacesAsync(cluster, antiAffinityGroup, tenant)
+ .thenAccept(asyncResponse::resume)
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to get all namespaces in cluster of given anti-affinity group, cluster: {}, "
+ + "tenant: {}, antiAffinityGroup: {}", clientAppId(), cluster, tenant, antiAffinityGroup,
+ ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@GET
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
index 2aaa5676d5b..2f5b99bfd9d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
@@ -2433,4 +2433,84 @@ public class NamespacesTest extends MockedPulsarServiceBaseTest {
setBookieAffinityGroupNs));
assertNull(bookieAffinityGroupDataResp);
}
+
+ @Test
+ public void testSetAndDeleteNamespaceAntiAffinityGroup() throws Exception {
+ // 1. create namespace with empty policies, namespace anti affinity group should be null
+ String setNamespaceAntiAffinityGroupNs = "test-set-namespace-anti-affinity-group-ns";
+ asyncRequests(response -> namespaces.createNamespace(response, testTenant, testLocalCluster,
+ setNamespaceAntiAffinityGroupNs, (Policies) null));
+ String namespaceAntiAffinityGroupResp = (String) asyncRequests(
+ response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant, testLocalCluster,
+ setNamespaceAntiAffinityGroupNs));
+ assertNull(namespaceAntiAffinityGroupResp);
+
+ // 2.set namespace anti affinity group
+ String namespaceAntiAffinityGroupReq = "namespace-anti-affinity-group";
+ asyncRequests(response -> namespaces.setNamespaceAntiAffinityGroup(response, testTenant, testLocalCluster,
+ setNamespaceAntiAffinityGroupNs, namespaceAntiAffinityGroupReq));
+
+ // 3.assert namespace anti affinity group
+ namespaceAntiAffinityGroupResp = (String) asyncRequests(
+ response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant, testLocalCluster,
+ setNamespaceAntiAffinityGroupNs));
+ assertEquals(namespaceAntiAffinityGroupResp, namespaceAntiAffinityGroupReq);
+
+ // 4.delete namespace anti affinity group
+ asyncRequests(response -> namespaces.removeNamespaceAntiAffinityGroup(response, testTenant, testLocalCluster,
+ setNamespaceAntiAffinityGroupNs));
+
+ // 5.assert namespace anti affinity group
+ namespaceAntiAffinityGroupResp = (String) asyncRequests(
+ response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant, testLocalCluster,
+ setNamespaceAntiAffinityGroupNs));
+ assertNull(namespaceAntiAffinityGroupResp);
+ }
+
+ @Test
+ public void testGetClusterAntiAffinityNamespaces() throws Exception {
+ // create 5 namespaces, 3 namespaces are set to the same namespace anti affinity group,
+ // 2 namespaces are not set to any anti affinity group
+ String namespaceWithAntiAffinity1 = "namespace-with-anti-affinity-1";
+ String namespaceWithAntiAffinity2 = "namespace-with-anti-affinity-2";
+ String namespaceWithAntiAffinity3 = "namespace-with-anti-affinity-3";
+ String namespaceWithoutAntiAffinity1 = "namespace-without-anti-affinity-1";
+ String namespaceWithoutAntiAffinity2 = "namespace-without-anti-affinity-2";
+
+ // create namespaces
+ List<String> allNamespaces =
+ List.of(namespaceWithAntiAffinity1, namespaceWithAntiAffinity2, namespaceWithAntiAffinity3,
+ namespaceWithoutAntiAffinity1, namespaceWithoutAntiAffinity2);
+ for (String namespace : allNamespaces) {
+ asyncRequests(response -> namespaces.createNamespace(response, testTenant, testLocalCluster, namespace,
+ (Policies) null));
+ }
+
+ // set namespace anti affinity group
+ String namespaceAntiAffinityGroupReq = "namespace-anti-affinity-group";
+ List<String> namespacesWithAntiAffinityGroup =
+ List.of(namespaceWithAntiAffinity1, namespaceWithAntiAffinity2, namespaceWithAntiAffinity3);
+ for (String namespace : namespacesWithAntiAffinityGroup) {
+ asyncRequests(response -> namespaces.setNamespaceAntiAffinityGroup(response, testTenant, testLocalCluster,
+ namespace, namespaceAntiAffinityGroupReq));
+ }
+
+ // assert namespace anti affinity group
+ for (String namespace : namespacesWithAntiAffinityGroup) {
+ String namespaceAntiAffinityGroupResp = (String) asyncRequests(
+ response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant, testLocalCluster,
+ namespace));
+ assertEquals(namespaceAntiAffinityGroupResp, namespaceAntiAffinityGroupReq);
+ }
+
+ // get namespaces in cluster of given anti affinity group
+ List<String> namespacesResp = (List<String>) asyncRequests(
+ response -> namespaces.getAntiAffinityNamespaces(response, testLocalCluster,
+ namespaceAntiAffinityGroupReq, testTenant));
+ List<String> namespacesWithFullPath =
+ namespacesWithAntiAffinityGroup.stream().map(ns -> NamespaceName.get(testTenant, testLocalCluster, ns))
+ .map(NamespaceName::toString).toList();
+ assertEquals(namespacesResp, namespacesWithFullPath);
+ }
+
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java
index 629e92c056f..596cfa3f396 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java
@@ -305,8 +305,8 @@ public class NamespacesV2Test extends MockedPulsarServiceBaseTest {
// 2.set namespace anti affinity group
String namespaceAntiAffinityGroupReq = "namespace-anti-affinity-group";
- namespaces.setNamespaceAntiAffinityGroup(testTenant, setNamespaceAntiAffinityGroupNs,
- namespaceAntiAffinityGroupReq);
+ asyncRequests(response -> namespaces.setNamespaceAntiAffinityGroup(response, testTenant,
+ setNamespaceAntiAffinityGroupNs, namespaceAntiAffinityGroupReq));
// 3.query namespace num bundles, should be conf.getDefaultNumberOfNamespaceBundles()
BundlesData bundlesData = (BundlesData) asyncRequests(
@@ -314,8 +314,9 @@ public class NamespacesV2Test extends MockedPulsarServiceBaseTest {
assertEquals(bundlesData.getNumBundles(), conf.getDefaultNumberOfNamespaceBundles());
// 4.assert namespace anti affinity group
- String namespaceAntiAffinityGroupResp =
- namespaces.getNamespaceAntiAffinityGroup(testTenant, setNamespaceAntiAffinityGroupNs);
+ String namespaceAntiAffinityGroupResp = (String) asyncRequests(
+ response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant,
+ setNamespaceAntiAffinityGroupNs));
assertEquals(namespaceAntiAffinityGroupResp, namespaceAntiAffinityGroupReq);
}
@@ -330,8 +331,8 @@ public class NamespacesV2Test extends MockedPulsarServiceBaseTest {
// 2.set namespace anti affinity group
String namespaceAntiAffinityGroupReq = "namespace-anti-affinity-group";
- namespaces.setNamespaceAntiAffinityGroup(testTenant, setNamespaceAntiAffinityGroupNs,
- namespaceAntiAffinityGroupReq);
+ asyncRequests(response -> namespaces.setNamespaceAntiAffinityGroup(response, testTenant,
+ setNamespaceAntiAffinityGroupNs, namespaceAntiAffinityGroupReq));
// 3.query namespace num bundles, should be policies.bundles, which we set before
BundlesData bundlesData = (BundlesData) asyncRequests(
@@ -339,8 +340,9 @@ public class NamespacesV2Test extends MockedPulsarServiceBaseTest {
assertEquals(bundlesData, policies.bundles);
// 4.assert namespace anti affinity group
- String namespaceAntiAffinityGroupResp =
- namespaces.getNamespaceAntiAffinityGroup(testTenant, setNamespaceAntiAffinityGroupNs);
+ String namespaceAntiAffinityGroupResp = (String) asyncRequests(
+ response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant,
+ setNamespaceAntiAffinityGroupNs));
assertEquals(namespaceAntiAffinityGroupResp, namespaceAntiAffinityGroupReq);
}
@@ -418,4 +420,81 @@ public class NamespacesV2Test extends MockedPulsarServiceBaseTest {
assertNull(bookieAffinityGroupDataResp);
}
+ @Test
+ public void testSetAndDeleteNamespaceAntiAffinityGroup() throws Exception {
+ // 1. create namespace with empty policies, namespace anti affinity group should be null
+ String setNamespaceAntiAffinityGroupNs = "test-set-namespace-anti-affinity-group-ns";
+ asyncRequests(
+ response -> namespaces.createNamespace(response, testTenant, setNamespaceAntiAffinityGroupNs, null));
+ String namespaceAntiAffinityGroupResp = (String) asyncRequests(
+ response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant,
+ setNamespaceAntiAffinityGroupNs));
+ assertNull(namespaceAntiAffinityGroupResp);
+
+ // 2.set namespace anti affinity group
+ String namespaceAntiAffinityGroupReq = "namespace-anti-affinity-group";
+ asyncRequests(response -> namespaces.setNamespaceAntiAffinityGroup(response, testTenant,
+ setNamespaceAntiAffinityGroupNs, namespaceAntiAffinityGroupReq));
+
+ // 3.assert namespace anti affinity group
+ namespaceAntiAffinityGroupResp = (String) asyncRequests(
+ response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant,
+ setNamespaceAntiAffinityGroupNs));
+ assertEquals(namespaceAntiAffinityGroupResp, namespaceAntiAffinityGroupReq);
+
+ // 4.delete namespace anti affinity group
+ asyncRequests(response -> namespaces.removeNamespaceAntiAffinityGroup(response, testTenant,
+ setNamespaceAntiAffinityGroupNs));
+
+ // 5.assert namespace anti affinity group
+ namespaceAntiAffinityGroupResp = (String) asyncRequests(
+ response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant,
+ setNamespaceAntiAffinityGroupNs));
+ assertNull(namespaceAntiAffinityGroupResp);
+ }
+
+ @Test
+ public void testGetClusterAntiAffinityNamespaces() throws Exception {
+ // create 5 namespaces, 3 namespaces are set to the same namespace anti affinity group,
+ // 2 namespaces are not set to any anti affinity group
+ String namespaceWithAntiAffinity1 = "namespace-with-anti-affinity-1";
+ String namespaceWithAntiAffinity2 = "namespace-with-anti-affinity-2";
+ String namespaceWithAntiAffinity3 = "namespace-with-anti-affinity-3";
+ String namespaceWithoutAntiAffinity1 = "namespace-without-anti-affinity-1";
+ String namespaceWithoutAntiAffinity2 = "namespace-without-anti-affinity-2";
+
+ // create namespaces
+ List<String> allNamespaces =
+ List.of(namespaceWithAntiAffinity1, namespaceWithAntiAffinity2, namespaceWithAntiAffinity3,
+ namespaceWithoutAntiAffinity1, namespaceWithoutAntiAffinity2);
+ for (String namespace : allNamespaces) {
+ asyncRequests(response -> namespaces.createNamespace(response, testTenant, namespace, null));
+ }
+
+ // set namespace anti affinity group
+ String namespaceAntiAffinityGroupReq = "namespace-anti-affinity-group";
+ List<String> namespacesWithAntiAffinityGroup =
+ List.of(namespaceWithAntiAffinity1, namespaceWithAntiAffinity2, namespaceWithAntiAffinity3);
+ for (String namespace : namespacesWithAntiAffinityGroup) {
+ asyncRequests(response -> namespaces.setNamespaceAntiAffinityGroup(response, testTenant, namespace,
+ namespaceAntiAffinityGroupReq));
+ }
+
+ // assert namespace anti affinity group
+ for (String namespace : namespacesWithAntiAffinityGroup) {
+ String namespaceAntiAffinityGroupResp = (String) asyncRequests(
+ response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant, namespace));
+ assertEquals(namespaceAntiAffinityGroupResp, namespaceAntiAffinityGroupReq);
+ }
+
+ // get namespaces in cluster of given anti affinity group
+ List<String> namespacesResp = (List<String>) asyncRequests(
+ response -> namespaces.getAntiAffinityNamespaces(response, testLocalCluster,
+ namespaceAntiAffinityGroupReq, testTenant));
+ List<String> namespacesWithFullPath =
+ namespacesWithAntiAffinityGroup.stream().map(ns -> NamespaceName.get(testTenant, ns))
+ .map(NamespaceName::toString).toList();
+ assertEquals(namespacesResp, namespacesWithFullPath);
+ }
+
}
(pulsar) 09/11: [fix][build][branch-4.0] Fix missing exclusion in cherry-picking #25264
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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit bad69d87095c221aa98f2afddd5bd0c156e3815f
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Wed Apr 15 18:52:03 2026 +0300
[fix][build][branch-4.0] Fix missing exclusion in cherry-picking #25264
---
pom.xml | 4 ++++
1 file changed, 4 insertions(+)
diff --git a/pom.xml b/pom.xml
index f835711b054..c88cdcec6ff 100644
--- a/pom.xml
+++ b/pom.xml
@@ -933,6 +933,10 @@ flexible messaging model and an intuitive client API.</description>
<groupId>jakarta.activation</groupId>
<artifactId>jakarta.activation-api</artifactId>
</exclusion>
+ <exclusion>
+ <groupId>javax.xml.bind</groupId>
+ <artifactId>jaxb-api</artifactId>
+ </exclusion>
</exclusions>
</dependency>
(pulsar) 07/11: [fix][admin] Refactor namespace migration operation to async in rest api (#25478)
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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b960c28983b99cf456cda95b82708079319f7516
Author: Oneby Wang <44...@users.noreply.github.com>
AuthorDate: Tue Apr 7 11:00:13 2026 +0800
[fix][admin] Refactor namespace migration operation to async in rest api (#25478)
(cherry picked from commit f72a5a78c8f705093bae68847dc1b7dbc38d7fb1)
---
.../pulsar/broker/admin/impl/NamespacesBase.java | 38 +++++-----------------
.../apache/pulsar/broker/admin/v1/Namespaces.java | 12 +++++--
.../apache/pulsar/broker/admin/v2/Namespaces.java | 12 +++++--
.../pulsar/broker/admin/NamespacesV2Test.java | 25 ++++++++++++--
4 files changed, 52 insertions(+), 35 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 bb4cac85087..56f4fc1400c 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
@@ -2801,22 +2801,15 @@ public abstract class NamespacesBase extends AdminResource {
});
}
- protected void internalEnableMigration(boolean migrated) {
- validateSuperUserAccess();
- try {
- getLocalPolicies().setLocalPoliciesWithCreate(namespaceName, oldPolicies -> oldPolicies.map(
- policies -> new LocalPolicies(policies.bundles,
- policies.bookieAffinityGroup,
- policies.namespaceAntiAffinityGroup,
- migrated))
- .orElseGet(() -> new LocalPolicies(getDefaultBundleData(), null, null, migrated)));
- log.info("Successfully updated migration on namespace {}", namespaceName);
- } catch (RestException re) {
- throw re;
- } catch (Exception e) {
- log.error("Failed to update migration on namespace {}", namespaceName, e);
- throw new RestException(e);
- }
+ protected CompletableFuture<Void> internalEnableMigrationAsync(boolean migrated) {
+ return validateSuperUserAccessAsync().thenCompose(__ -> getDefaultBundleDataAsync().thenCompose(
+ defaultBundleData -> getLocalPolicies().setLocalPoliciesWithCreateAsync(namespaceName,
+ oldPolicies -> oldPolicies.map(
+ policies -> new LocalPolicies(policies.bundles,
+ policies.bookieAffinityGroup,
+ policies.namespaceAntiAffinityGroup, migrated))
+ .orElseGet(() -> new LocalPolicies(defaultBundleData, null, null, migrated)))))
+ .thenAccept(__ -> log.info("Successfully updated migration on namespace {}", namespaceName));
}
protected Policies getDefaultPolicesIfNull(Policies policies) {
@@ -2914,19 +2907,6 @@ public abstract class NamespacesBase extends AdminResource {
.thenApply(policies -> policies.allowed_clusters);
}
- // TODO remove this sync method after async refactor
- @Deprecated
- private BundlesData getDefaultBundleData() {
- try {
- return getDefaultBundleDataAsync().get(config().getMetadataStoreOperationTimeoutSeconds(),
- TimeUnit.SECONDS);
- } catch (Exception e) {
- log.error("[{}] Failed to get namespace-policy configuration for namespace {}", clientAppId(),
- namespaceName, e);
- throw new RestException(e);
- }
- }
-
private CompletableFuture<BundlesData> getDefaultBundleDataAsync() {
return namespaceResources().getPoliciesAsync(namespaceName).thenApply(
optionalPolicies -> optionalPolicies.isPresent() ? optionalPolicies.get().bundles :
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 4ba8d6aa1da..1c736d749ee 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
@@ -1781,12 +1781,20 @@ public class Namespaces extends NamespacesBase {
@ApiOperation(hidden = true, value = "Update migration for all topics in a namespace")
@ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Property or cluster or namespace doesn't exist") })
- public void enableMigration(@PathParam("property") String property,
+ public void enableMigration(@Suspended AsyncResponse asyncResponse,
+ @PathParam("property") String property,
@PathParam("cluster") String cluster,
@PathParam("namespace") String namespace,
boolean migrated) {
validateNamespaceName(property, cluster, namespace);
- internalEnableMigration(migrated);
+ internalEnableMigrationAsync(migrated)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to update migration, tenant: {}, namespace: {}",
+ clientAppId(), property, namespace, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@PUT
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 732cb315604..836d2699e18 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
@@ -3083,11 +3083,19 @@ public class Namespaces extends NamespacesBase {
@ApiResponse(code = 200, message = "Operation successful"),
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Property or cluster or namespace doesn't exist") })
- public void enableMigration(@PathParam("tenant") String tenant,
+ public void enableMigration(@Suspended AsyncResponse asyncResponse,
+ @PathParam("tenant") String tenant,
@PathParam("namespace") String namespace,
boolean migrated) {
validateNamespaceName(tenant, namespace);
- internalEnableMigration(migrated);
+ internalEnableMigrationAsync(migrated)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to update migration, tenant: {}, namespace: {}",
+ clientAppId(), tenant, namespace, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@POST
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java
index 596cfa3f396..44523a16435 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java
@@ -24,6 +24,7 @@ import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
@@ -354,7 +355,8 @@ public class NamespacesV2Test extends MockedPulsarServiceBaseTest {
// 2.set enable migration
boolean enableMigrationReq = true;
- namespaces.enableMigration(testTenant, enableMigrationGroupNs, enableMigrationReq);
+ asyncRequests(response -> namespaces.enableMigration(response, testTenant, enableMigrationGroupNs,
+ enableMigrationReq));
// 3.query namespace num bundles, should be conf.getDefaultNumberOfNamespaceBundles()
BundlesData bundlesData = (BundlesData) asyncRequests(
@@ -378,7 +380,8 @@ public class NamespacesV2Test extends MockedPulsarServiceBaseTest {
// 2.set enable migration
boolean enableMigrationReq = true;
- namespaces.enableMigration(testTenant, enableMigrationGroupNs, enableMigrationReq);
+ asyncRequests(response -> namespaces.enableMigration(response, testTenant, enableMigrationGroupNs,
+ enableMigrationReq));
// 3.query namespace num bundles, should be policies.bundles, which we set before
BundlesData bundlesData = (BundlesData) asyncRequests(
@@ -497,4 +500,22 @@ public class NamespacesV2Test extends MockedPulsarServiceBaseTest {
assertEquals(namespacesResp, namespacesWithFullPath);
}
+ @Test
+ public void testEnableMigrationAndDisableMigration() throws Exception {
+ String enableMigrationGroupNs = "test-enable-migration-disable-migration-ns";
+ asyncRequests(response -> namespaces.createNamespace(response, testTenant, enableMigrationGroupNs, null));
+
+ // Enable migration
+ asyncRequests(response -> namespaces.enableMigration(response, testTenant, enableMigrationGroupNs, true));
+ Policies policiesResp = (Policies) asyncRequests(
+ response -> namespaces.getPolicies(response, testTenant, enableMigrationGroupNs));
+ assertTrue(policiesResp.migrated);
+
+ // Disable migration
+ asyncRequests(response -> namespaces.enableMigration(response, testTenant, enableMigrationGroupNs, false));
+ policiesResp = (Policies) asyncRequests(
+ response -> namespaces.getPolicies(response, testTenant, enableMigrationGroupNs));
+ assertFalse(policiesResp.migrated);
+ }
+
}
(pulsar) 01/11: [fix][build][branch-4.0] Fix broken compilation after cherry-picking #25400
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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f0e98a20bbc8657e569e0d588afcc28185e232ef
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Mar 31 23:20:40 2026 +0300
[fix][build][branch-4.0] Fix broken compilation after cherry-picking #25400
---
.../src/main/java/org/apache/pulsar/common/util/SecurityUtility.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/SecurityUtility.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/SecurityUtility.java
index fa72327dc47..80c692c9d6f 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/SecurityUtility.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/SecurityUtility.java
@@ -52,6 +52,7 @@ import java.security.spec.InvalidKeySpecException;
import java.security.spec.KeySpec;
import java.security.spec.PKCS8EncodedKeySpec;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Base64;
import java.util.Collection;
import java.util.List;
@@ -82,7 +83,7 @@ public class SecurityUtility {
public static final String BC_NON_FIPS_PROVIDER_CLASS = "org.bouncycastle.jce.provider.BouncyCastleProvider";
public static final String CONSCRYPT_PROVIDER_CLASS = "org.conscrypt.OpenSSLProvider";
public static final Provider CONSCRYPT_PROVIDER = loadConscryptProvider();
- private static final List<String> KEY_FACTORY_ALGORITHMS = List.of("RSA", "EC");
+ private static final List<String> KEY_FACTORY_ALGORITHMS = Arrays.asList("RSA", "EC");
// Security.getProvider("BC") / Security.getProvider("BCFIPS").
// also used to get Factories. e.g. CertificateFactory.getInstance("X.509", "BCFIPS")
(pulsar) 03/11: Revert "[fix][broker] Fix backlog clearing for unloaded namespace bundles (#25272)"
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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit bc8449deb3adcf4b5e8ba39eb01d2fd3e5743f44
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Wed Apr 15 17:34:26 2026 +0300
Revert "[fix][broker] Fix backlog clearing for unloaded namespace bundles (#25272)"
This reverts commit 24e7814cbd49a94007456b6be2492bc8c1881b5e.
---
.../pulsar/broker/admin/impl/NamespacesBase.java | 273 +++++++++++----------
.../apache/pulsar/broker/admin/v2/Namespaces.java | 57 ++---
.../pulsar/broker/namespace/NamespaceService.java | 19 --
.../apache/pulsar/broker/admin/AdminApiTest.java | 126 ----------
4 files changed, 166 insertions(+), 309 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 8e1d11b15f3..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
@@ -79,6 +79,7 @@ import org.apache.pulsar.common.naming.NamedEntity;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceBundleFactory;
import org.apache.pulsar.common.naming.NamespaceBundleSplitAlgorithm;
+import org.apache.pulsar.common.naming.NamespaceBundles;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.SystemTopicNames;
import org.apache.pulsar.common.naming.TopicName;
@@ -1567,108 +1568,139 @@ public abstract class NamespacesBase extends AdminResource {
);
}
- protected CompletableFuture<Void> internalClearNamespaceBacklogAsync(boolean authoritative) {
- return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.CLEAR_BACKLOG)
- .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()
- .clearNamespaceBundleBacklogAsync(namespaceName.toString(),
- nsBundle.getBundleRange()));
- } catch (PulsarServerException e) {
- return CompletableFuture.failedFuture(e);
- }
- }
- return FutureUtil.waitForAll(futures);
- }).thenRun(() -> log.info("[{}] Successfully cleared backlog on all the bundles for namespace {}",
- clientAppId(), namespaceName));
+ protected void internalClearNamespaceBacklog(AsyncResponse asyncResponse, boolean authoritative) {
+ validateNamespaceOperation(namespaceName, NamespaceOperation.CLEAR_BACKLOG);
+
+ 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 is no backlog on this bundle to clear
+ if (pulsar().getNamespaceService().checkOwnershipPresent(nsBundle)) {
+ futures.add(pulsar().getAdminClient().namespaces()
+ .clearNamespaceBundleBacklogAsync(namespaceName.toString(), nsBundle.getBundleRange()));
+ }
+ }
+ } 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 clear backlog on the bundles for namespace {}: {}", clientAppId(),
+ 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 cleared backlog on all the bundles for namespace {}", clientAppId(),
+ namespaceName);
+ asyncResponse.resume(Response.noContent().build());
+ return null;
+ });
}
@SuppressWarnings("deprecation")
- protected CompletableFuture<Void> internalClearNamespaceBundleBacklogAsync(String bundleRange,
- boolean authoritative) {
- if (bundleRange == null) {
- return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "BundleRange should not be null"));
+ protected void internalClearNamespaceBundleBacklog(String bundleRange, boolean authoritative) {
+ validateNamespaceOperation(namespaceName, NamespaceOperation.CLEAR_BACKLOG);
+ 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());
}
- return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.CLEAR_BACKLOG)
- .thenCompose(__ -> {
- // check cluster ownership for a given global namespace: redirect if peer-cluster owns it
- return validateGlobalNamespaceOwnershipAsync(namespaceName);
- })
- .thenCompose(__ -> getNamespacePoliciesAsync(namespaceName))
- .thenCompose(policies ->
- // Allow acquiring ownership for an unassigned bundle so backlog can be cleared
- // even if not loaded.
- validateNamespaceBundleOwnershipAsync(namespaceName, policies.bundles, bundleRange,
- authoritative, false))
- .thenCompose(bundle -> clearBacklogAsync(bundle, null))
- .thenRun(() -> log.info("[{}] Successfully cleared backlog on namespace bundle {}/{}", clientAppId(),
- namespaceName, bundleRange));
+ validateNamespaceBundleOwnership(namespaceName, policies.bundles, bundleRange, authoritative, true);
+
+ clearBacklog(namespaceName, bundleRange, null);
+ log.info("[{}] Successfully cleared backlog on namespace bundle {}/{}", clientAppId(), namespaceName,
+ bundleRange);
}
- protected CompletableFuture<Void> internalClearNamespaceBacklogForSubscriptionAsync(String subscription,
- boolean authoritative) {
- if (subscription == null) {
- return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "Subscription should not be null"));
+ protected void internalClearNamespaceBacklogForSubscription(AsyncResponse asyncResponse, String subscription,
+ boolean authoritative) {
+ validateNamespaceOperation(namespaceName, NamespaceOperation.CLEAR_BACKLOG);
+ 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 is no backlog on this bundle to clear
+ if (pulsar().getNamespaceService().checkOwnershipPresent(nsBundle)) {
+ futures.add(pulsar().getAdminClient().namespaces().clearNamespaceBundleBacklogForSubscriptionAsync(
+ namespaceName.toString(), nsBundle.getBundleRange(), subscription));
+ }
+ }
+ } catch (WebApplicationException wae) {
+ asyncResponse.resume(wae);
+ return;
+ } catch (Exception e) {
+ asyncResponse.resume(new RestException(e));
+ return;
}
- return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.CLEAR_BACKLOG)
- .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()
- .clearNamespaceBundleBacklogForSubscriptionAsync(
- namespaceName.toString(), nsBundle.getBundleRange(), subscription));
- } catch (PulsarServerException e) {
- return CompletableFuture.failedFuture(e);
- }
- }
- return FutureUtil.waitForAll(futures);
- }).thenRun(() -> log.info(
- "[{}] Successfully cleared backlog for subscription {} on all the bundles for namespace {}",
- clientAppId(), subscription, namespaceName));
+ FutureUtil.waitForAll(futures).handle((result, exception) -> {
+ if (exception != null) {
+ log.warn("[{}] Failed to clear backlog for subscription {} 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 cleared backlog for subscription {} on all the bundles for namespace {}",
+ clientAppId(), subscription, namespaceName);
+ asyncResponse.resume(Response.noContent().build());
+ return null;
+ });
}
@SuppressWarnings("deprecation")
- protected CompletableFuture<Void> internalClearNamespaceBundleBacklogForSubscriptionAsync(String subscription,
- String bundleRange,
- boolean authoritative) {
- if (subscription == null) {
- return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "Subscription should not be null"));
- }
- if (bundleRange == null) {
- return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "BundleRange should not be null"));
+ protected void internalClearNamespaceBundleBacklogForSubscription(String subscription, String bundleRange,
+ boolean authoritative) {
+ validateNamespaceOperation(namespaceName, NamespaceOperation.CLEAR_BACKLOG);
+ 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());
}
- return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.CLEAR_BACKLOG)
- .thenCompose(__ -> {
- // check cluster ownership for a given global namespace: redirect if peer-cluster owns it
- return validateGlobalNamespaceOwnershipAsync(namespaceName);
- })
- .thenCompose(__ -> getNamespacePoliciesAsync(namespaceName))
- .thenCompose(policies ->
- // Allow acquiring ownership for an unassigned bundle so backlog can be cleared
- // even if not loaded.
- validateNamespaceBundleOwnershipAsync(namespaceName, policies.bundles, bundleRange,
- authoritative, false))
- .thenCompose(bundle -> clearBacklogAsync(bundle, subscription))
- .thenRun(() -> log.info(
- "[{}] Successfully cleared backlog for subscription {} on namespace bundle {}/{}",
- clientAppId(), subscription, namespaceName, bundleRange));
+ validateNamespaceBundleOwnership(namespaceName, policies.bundles, bundleRange, authoritative, true);
+
+ clearBacklog(namespaceName, bundleRange, subscription);
+ log.info("[{}] Successfully cleared backlog for subscription {} on namespace bundle {}/{}", clientAppId(),
+ subscription, namespaceName, bundleRange);
}
protected CompletableFuture<Void> internalUnsubscribeNamespaceAsync(String subscription,
boolean authoritative) {
- if (subscription == null) {
- return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "Subscription should not be null"));
- }
+ checkNotNull(subscription, "Subscription should not be null");
return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.UNSUBSCRIBE)
.thenCompose(__ -> pulsar().getNamespaceService().getNamespaceBundleFactory()
@@ -1691,12 +1723,8 @@ public abstract class NamespacesBase extends AdminResource {
@SuppressWarnings("deprecation")
protected CompletableFuture<Void> internalUnsubscribeNamespaceBundleAsync(String subscription, String bundleRange,
boolean authoritative) {
- if (subscription == null) {
- return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "Subscription should not be null"));
- }
- if (bundleRange == null) {
- return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "BundleRange should not be null"));
- }
+ checkNotNull(subscription, "Subscription should not be null");
+ checkNotNull(bundleRange, "BundleRange should not be null");
return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.UNSUBSCRIBE)
.thenCompose(__ -> {
@@ -1904,47 +1932,38 @@ public abstract class NamespacesBase extends AdminResource {
return checkBacklogQuota(quota, retention);
}
- private CompletableFuture<Void> clearBacklogAsync(NamespaceBundle bundle, String subscription) {
- return pulsar().getNamespaceService().getOwnedPersistentTopicListForNamespaceBundle(bundle)
- .thenCompose(topicsInBundle -> {
- List<CompletableFuture<Void>> futures = new ArrayList<>();
- String effectiveSubscription = subscription;
- if (effectiveSubscription != null
- && effectiveSubscription.startsWith(pulsar().getConfiguration().getReplicatorPrefix())) {
- effectiveSubscription = PersistentReplicator.getRemoteCluster(effectiveSubscription);
- }
- final String finalSubscription = effectiveSubscription;
+ private void clearBacklog(NamespaceName nsName, String bundleRange, String subscription) {
+ try {
+ List<Topic> topicList = pulsar().getBrokerService().getAllTopicsFromNamespaceBundle(nsName.toString(),
+ nsName.toString() + "/" + bundleRange);
- 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();
- if (!(loaded instanceof PersistentTopic persistentTopic)) {
- return CompletableFuture.completedFuture(null);
- }
- return finalSubscription != null
- ? persistentTopic.clearBacklog(finalSubscription)
- : persistentTopic.clearBacklog();
- }));
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ if (subscription != null) {
+ if (subscription.startsWith(pulsar().getConfiguration().getReplicatorPrefix())) {
+ subscription = PersistentReplicator.getRemoteCluster(subscription);
+ }
+ for (Topic topic : topicList) {
+ if (topic instanceof PersistentTopic
+ && !pulsar().getBrokerService().isSystemTopic(TopicName.get(topic.getName()))) {
+ futures.add(((PersistentTopic) topic).clearBacklog(subscription));
}
-
- return FutureUtil.waitForAll(futures);
- }).exceptionally(ex -> {
- Throwable cause = FutureUtil.unwrapCompletionException(ex);
- if (cause instanceof RestException) {
- throw (RestException) cause;
+ }
+ } else {
+ for (Topic topic : topicList) {
+ if (topic instanceof PersistentTopic
+ && !pulsar().getBrokerService().isSystemTopic(TopicName.get(topic.getName()))) {
+ futures.add(((PersistentTopic) topic).clearBacklog());
}
- throw new RestException(cause);
- });
- }
+ }
+ }
+ FutureUtil.waitForAll(futures).get();
+ } catch (Exception e) {
+ log.error("[{}] Failed to clear backlog for namespace {}/{}, subscription: {}", clientAppId(),
+ nsName.toString(), bundleRange, subscription, e);
+ throw new RestException(e);
+ }
+ }
private CompletableFuture<Void> unsubscribeAsync(NamespaceBundle bundle, String subscription) {
if (subscription.startsWith(pulsar().getConfiguration().getReplicatorPrefix())) {
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 a997e1d898a..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
@@ -1512,14 +1512,14 @@ public class Namespaces extends NamespacesBase {
public void clearNamespaceBacklog(@Suspended final AsyncResponse asyncResponse, @PathParam("tenant") String tenant,
@PathParam("namespace") String namespace,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
- validateNamespaceName(tenant, namespace);
- internalClearNamespaceBacklogAsync(authoritative)
- .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
- .exceptionally(ex -> {
- log.error("[{}] Failed to clear backlog on namespace {}", clientAppId(), namespaceName, ex);
- resumeAsyncResponseExceptionally(asyncResponse, ex);
- return null;
- });
+ try {
+ validateNamespaceName(tenant, namespace);
+ internalClearNamespaceBacklog(asyncResponse, authoritative);
+ } catch (WebApplicationException wae) {
+ asyncResponse.resume(wae);
+ } catch (Exception e) {
+ asyncResponse.resume(new RestException(e));
+ }
}
@POST
@@ -1530,19 +1530,11 @@ public class Namespaces extends NamespacesBase {
@ApiResponse(code = 307, message = "Current broker doesn't serve the namespace"),
@ApiResponse(code = 403, message = "Don't have admin or operate permission on the namespace"),
@ApiResponse(code = 404, message = "Namespace does not exist") })
- public void clearNamespaceBundleBacklog(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
+ public void clearNamespaceBundleBacklog(@PathParam("tenant") String tenant,
@PathParam("namespace") String namespace, @PathParam("bundle") String bundleRange,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateNamespaceName(tenant, namespace);
- internalClearNamespaceBundleBacklogAsync(bundleRange, authoritative)
- .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
- .exceptionally(ex -> {
- log.error("[{}] Failed to clear backlog on namespace bundle {}/{}", clientAppId(),
- namespaceName, bundleRange, ex);
- resumeAsyncResponseExceptionally(asyncResponse, ex);
- return null;
- });
+ internalClearNamespaceBundleBacklog(bundleRange, authoritative);
}
@POST
@@ -1556,15 +1548,14 @@ public class Namespaces extends NamespacesBase {
@PathParam("tenant") String tenant, @PathParam("namespace") String namespace,
@PathParam("subscription") String subscription,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
- validateNamespaceName(tenant, namespace);
- internalClearNamespaceBacklogForSubscriptionAsync(subscription, authoritative)
- .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
- .exceptionally(ex -> {
- log.error("[{}] Failed to clear backlog for subscription {} on namespace {}", clientAppId(),
- subscription, namespaceName, ex);
- resumeAsyncResponseExceptionally(asyncResponse, ex);
- return null;
- });
+ try {
+ validateNamespaceName(tenant, namespace);
+ internalClearNamespaceBacklogForSubscription(asyncResponse, subscription, authoritative);
+ } catch (WebApplicationException wae) {
+ asyncResponse.resume(wae);
+ } catch (Exception e) {
+ asyncResponse.resume(new RestException(e));
+ }
}
@POST
@@ -1575,20 +1566,12 @@ public class Namespaces extends NamespacesBase {
@ApiResponse(code = 307, message = "Current broker doesn't serve the namespace"),
@ApiResponse(code = 403, message = "Don't have admin or operate permission on the namespace"),
@ApiResponse(code = 404, message = "Namespace does not exist") })
- public void clearNamespaceBundleBacklogForSubscription(@Suspended final AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
+ public void clearNamespaceBundleBacklogForSubscription(@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);
- internalClearNamespaceBundleBacklogForSubscriptionAsync(subscription, bundleRange, authoritative)
- .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
- .exceptionally(ex -> {
- log.error("[{}] Failed to clear backlog for subscription {} on namespace bundle {}/{}",
- clientAppId(), subscription, namespaceName, bundleRange, ex);
- resumeAsyncResponseExceptionally(asyncResponse, ex);
- return null;
- });
+ internalClearNamespaceBundleBacklogForSubscription(subscription, bundleRange, authoritative);
}
@POST
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index cda619b7018..b5dbddc2ea3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -1380,25 +1380,6 @@ public class NamespaceService implements AutoCloseable {
});
}
- public CompletableFuture<List<String>> getOwnedPersistentTopicListForNamespaceBundle(NamespaceBundle bundle) {
- return getListOfPersistentTopics(bundle.getNamespaceObject()).thenCompose(topics ->
- CompletableFuture.completedFuture(
- topics.stream()
- .filter(topic -> bundle.includes(TopicName.get(topic)))
- .collect(Collectors.toList())))
- .thenCombine(getPartitions(bundle.getNamespaceObject(), TopicDomain.persistent).thenCompose(topics ->
- CompletableFuture.completedFuture(
- topics.stream().filter(topic -> bundle.includes(TopicName.get(topic)))
- .collect(Collectors.toList()))), (left, right) -> {
- for (String topic : right) {
- if (!left.contains(topic)) {
- left.add(topic);
- }
- }
- return left;
- });
- }
-
/***
* Checks whether the topic exists( partitioned or non-partitioned ).
*/
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 9a5c72ff240..2c5d045f758 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
@@ -2284,132 +2284,6 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
assertEquals(backlog, 0);
}
- @Test(dataProvider = "numBundles")
- public void testClearNamespaceBundleBacklogOnUnloadedBundle(Integer numBundles) throws Exception {
- String namespace = "prop-xyz/ns1-bundles";
- admin.namespaces().createNamespace("prop-xyz/ns1-bundles", numBundles);
-
- String topic = "persistent://" + namespace + "/t1";
- String subscription = "sub1";
-
- Consumer<byte[]> consumer = pulsarClient.newConsumer()
- .topic(topic)
- .subscriptionName(subscription)
- .subscribe();
-
- Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
- .topic(topic)
- .enableBatching(false)
- .messageRoutingMode(MessageRoutingMode.SinglePartition)
- .create();
- for (int i = 0; i < 10; i++) {
- producer.send(("message-" + i).getBytes());
- }
- producer.close();
- consumer.close();
-
- long backlog = admin.topics().getStats(topic).getSubscriptions().get(subscription).getMsgBacklog();
- assertEquals(backlog, 10);
-
- 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().clearNamespaceBundleBacklog(namespace, bundle.getBundleRange());
-
- @Cleanup
- Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
- .topic(topic)
- .subscriptionName(subscription)
- .subscribe();
-
- long backlogAfter = admin.topics().getStats(topic).getSubscriptions().get(subscription).getMsgBacklog();
- assertEquals(backlogAfter, 0);
- }
-
- @Test(dataProvider = "numBundles")
- public void testClearNamespaceBundleBacklogForSubscriptionOnUnloadedBundle(Integer numBundles) throws Exception {
- String namespace = "prop-xyz/ns1-bundles";
- admin.namespaces().createNamespace("prop-xyz/ns1-bundles", 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();
-
- Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
- .topic(topic)
- .enableBatching(false)
- .messageRoutingMode(MessageRoutingMode.SinglePartition)
- .create();
- for (int i = 0; i < 10; i++) {
- producer.send(("message-" + i).getBytes());
- }
- producer.close();
- consumer1.close();
- consumer2.close();
-
- long backlog = admin.topics().getStats(topic).getSubscriptions().get(subscription).getMsgBacklog();
- assertEquals(backlog, 10);
- long otherBacklog = admin.topics().getStats(topic).getSubscriptions().get(otherSubscription).getMsgBacklog();
- assertEquals(otherBacklog, 10);
-
- 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().clearNamespaceBundleBacklogForSubscription(namespace, bundle.getBundleRange(),
- subscription);
-
- @Cleanup
- Consumer<byte[]> consumer3 = pulsarClient.newConsumer()
- .topic(topic)
- .subscriptionName(subscription)
- .subscribe();
-
- @Cleanup
- Consumer<byte[]> consumer4 = pulsarClient.newConsumer()
- .topic(topic)
- .subscriptionName(otherSubscription)
- .subscribe();
-
- long backlogAfter = admin.topics().getStats(topic).getSubscriptions().get(subscription).getMsgBacklog();
- assertEquals(backlogAfter, 0);
- long otherBacklogAfter =
- admin.topics().getStats(topic).getSubscriptions().get(otherSubscription).getMsgBacklog();
- assertEquals(otherBacklogAfter, 10);
- }
-
-
@Test(dataProvider = "bundling")
public void testUnsubscribeOnNamespace(Integer numBundles) throws Exception {
admin.namespaces().createNamespace("prop-xyz/ns1-bundles", numBundles);
(pulsar) 08/11: [fix][broker] Fix backlog clearing for unloaded namespace bundles (#25272)
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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit eb79b2d013028e3144aaa2feb34f105283e1ebae
Author: Hao Zhang <zh...@cmss.chinamobile.com>
AuthorDate: Tue Apr 7 11:36:30 2026 +0800
[fix][broker] Fix backlog clearing for unloaded namespace bundles (#25272)
Co-authored-by: 张浩 <zh...@100.me>
(cherry picked from commit 8e37e3ee63d8fcebed493daa6e577fdc08591746)
---
.../apache/pulsar/broker/admin/AdminResource.java | 7 -
.../pulsar/broker/admin/impl/NamespacesBase.java | 302 ++++++++++-----------
.../apache/pulsar/broker/admin/v1/Namespaces.java | 45 ++-
.../apache/pulsar/broker/admin/v2/Namespaces.java | 57 ++--
.../pulsar/broker/namespace/NamespaceService.java | 19 ++
.../apache/pulsar/broker/admin/AdminApiTest.java | 126 +++++++++
6 files changed, 370 insertions(+), 186 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
index 103181a7c73..dd45ac3c94c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
@@ -18,7 +18,6 @@
*/
package org.apache.pulsar.broker.admin;
-import static org.apache.commons.lang3.StringUtils.isBlank;
import com.fasterxml.jackson.databind.ObjectReader;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.google.errorprone.annotations.CanIgnoreReturnValue;
@@ -852,12 +851,6 @@ public abstract class AdminResource extends PulsarWebResource {
}
}
- protected void checkNotBlank(String str, String errorMessage) {
- if (isBlank(str)) {
- throw new RestException(Status.PRECONDITION_FAILED, errorMessage);
- }
- }
-
protected boolean isManagedLedgerNotFoundException(Throwable cause) {
return cause instanceof ManagedLedgerException.MetadataNotFoundException
|| cause instanceof MetadataStoreException.NotFoundException;
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 56f4fc1400c..b0006e459bb 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
@@ -70,7 +70,6 @@ import org.apache.pulsar.broker.topiclistlimit.TopicListSizeResultCache;
import org.apache.pulsar.broker.web.RestException;
import org.apache.pulsar.client.admin.GrantTopicPermissionOptions;
import org.apache.pulsar.client.admin.PulsarAdmin;
-import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.admin.RevokeTopicPermissionOptions;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace;
@@ -78,7 +77,6 @@ import org.apache.pulsar.common.naming.NamedEntity;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceBundleFactory;
import org.apache.pulsar.common.naming.NamespaceBundleSplitAlgorithm;
-import org.apache.pulsar.common.naming.NamespaceBundles;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.SystemTopicNames;
import org.apache.pulsar.common.naming.TopicName;
@@ -1538,139 +1536,108 @@ public abstract class NamespacesBase extends AdminResource {
);
}
- protected void internalClearNamespaceBacklog(AsyncResponse asyncResponse, boolean authoritative) {
- validateNamespaceOperation(namespaceName, NamespaceOperation.CLEAR_BACKLOG);
-
- 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 is no backlog on this bundle to clear
- if (pulsar().getNamespaceService().checkOwnershipPresent(nsBundle)) {
- futures.add(pulsar().getAdminClient().namespaces()
- .clearNamespaceBundleBacklogAsync(namespaceName.toString(), nsBundle.getBundleRange()));
- }
- }
- } 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 clear backlog on the bundles for namespace {}: {}", clientAppId(),
- 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 cleared backlog on all the bundles for namespace {}", clientAppId(),
- namespaceName);
- asyncResponse.resume(Response.noContent().build());
- return null;
- });
+ protected CompletableFuture<Void> internalClearNamespaceBacklogAsync(boolean authoritative) {
+ return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.CLEAR_BACKLOG)
+ .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()
+ .clearNamespaceBundleBacklogAsync(namespaceName.toString(),
+ nsBundle.getBundleRange()));
+ } catch (PulsarServerException e) {
+ return CompletableFuture.failedFuture(e);
+ }
+ }
+ return FutureUtil.waitForAll(futures);
+ }).thenRun(() -> log.info("[{}] Successfully cleared backlog on all the bundles for namespace {}",
+ clientAppId(), namespaceName));
}
@SuppressWarnings("deprecation")
- protected void internalClearNamespaceBundleBacklog(String bundleRange, boolean authoritative) {
- validateNamespaceOperation(namespaceName, NamespaceOperation.CLEAR_BACKLOG);
- 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());
+ protected CompletableFuture<Void> internalClearNamespaceBundleBacklogAsync(String bundleRange,
+ boolean authoritative) {
+ if (bundleRange == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "BundleRange should not be null"));
}
- validateNamespaceBundleOwnership(namespaceName, policies.bundles, bundleRange, authoritative, true);
-
- clearBacklog(namespaceName, bundleRange, null);
- log.info("[{}] Successfully cleared backlog on namespace bundle {}/{}", clientAppId(), namespaceName,
- bundleRange);
+ return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.CLEAR_BACKLOG)
+ .thenCompose(__ -> {
+ // check cluster ownership for a given global namespace: redirect if peer-cluster owns it
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ })
+ .thenCompose(__ -> getNamespacePoliciesAsync(namespaceName))
+ .thenCompose(policies ->
+ // Allow acquiring ownership for an unassigned bundle so backlog can be cleared
+ // even if not loaded.
+ validateNamespaceBundleOwnershipAsync(namespaceName, policies.bundles, bundleRange,
+ authoritative, false))
+ .thenCompose(bundle -> clearBacklogAsync(bundle, null))
+ .thenRun(() -> log.info("[{}] Successfully cleared backlog on namespace bundle {}/{}", clientAppId(),
+ namespaceName, bundleRange));
}
- protected void internalClearNamespaceBacklogForSubscription(AsyncResponse asyncResponse, String subscription,
- boolean authoritative) {
- validateNamespaceOperation(namespaceName, NamespaceOperation.CLEAR_BACKLOG);
- 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 is no backlog on this bundle to clear
- if (pulsar().getNamespaceService().checkOwnershipPresent(nsBundle)) {
- futures.add(pulsar().getAdminClient().namespaces().clearNamespaceBundleBacklogForSubscriptionAsync(
- namespaceName.toString(), nsBundle.getBundleRange(), subscription));
- }
- }
- } catch (WebApplicationException wae) {
- asyncResponse.resume(wae);
- return;
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- return;
+ protected CompletableFuture<Void> internalClearNamespaceBacklogForSubscriptionAsync(String subscription,
+ boolean authoritative) {
+ if (subscription == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "Subscription should not be null"));
}
- FutureUtil.waitForAll(futures).handle((result, exception) -> {
- if (exception != null) {
- log.warn("[{}] Failed to clear backlog for subscription {} 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 cleared backlog for subscription {} on all the bundles for namespace {}",
- clientAppId(), subscription, namespaceName);
- asyncResponse.resume(Response.noContent().build());
- return null;
- });
+ return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.CLEAR_BACKLOG)
+ .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()
+ .clearNamespaceBundleBacklogForSubscriptionAsync(
+ namespaceName.toString(), nsBundle.getBundleRange(), subscription));
+ } catch (PulsarServerException e) {
+ return CompletableFuture.failedFuture(e);
+ }
+ }
+ return FutureUtil.waitForAll(futures);
+ }).thenRun(() -> log.info(
+ "[{}] Successfully cleared backlog for subscription {} on all the bundles for namespace {}",
+ clientAppId(), subscription, namespaceName));
}
@SuppressWarnings("deprecation")
- protected void internalClearNamespaceBundleBacklogForSubscription(String subscription, String bundleRange,
- boolean authoritative) {
- validateNamespaceOperation(namespaceName, NamespaceOperation.CLEAR_BACKLOG);
- 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());
+ protected CompletableFuture<Void> internalClearNamespaceBundleBacklogForSubscriptionAsync(String subscription,
+ String bundleRange,
+ boolean authoritative) {
+ if (subscription == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "Subscription should not be null"));
+ }
+ if (bundleRange == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "BundleRange should not be null"));
}
- validateNamespaceBundleOwnership(namespaceName, policies.bundles, bundleRange, authoritative, true);
-
- clearBacklog(namespaceName, bundleRange, subscription);
- log.info("[{}] Successfully cleared backlog for subscription {} on namespace bundle {}/{}", clientAppId(),
- subscription, namespaceName, bundleRange);
+ return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.CLEAR_BACKLOG)
+ .thenCompose(__ -> {
+ // check cluster ownership for a given global namespace: redirect if peer-cluster owns it
+ return validateGlobalNamespaceOwnershipAsync(namespaceName);
+ })
+ .thenCompose(__ -> getNamespacePoliciesAsync(namespaceName))
+ .thenCompose(policies ->
+ // Allow acquiring ownership for an unassigned bundle so backlog can be cleared
+ // even if not loaded.
+ validateNamespaceBundleOwnershipAsync(namespaceName, policies.bundles, bundleRange,
+ authoritative, false))
+ .thenCompose(bundle -> clearBacklogAsync(bundle, subscription))
+ .thenRun(() -> log.info(
+ "[{}] Successfully cleared backlog for subscription {} on namespace bundle {}/{}",
+ clientAppId(), subscription, namespaceName, bundleRange));
}
protected CompletableFuture<Void> internalUnsubscribeNamespaceAsync(String subscription,
boolean authoritative) {
- checkNotNull(subscription, "Subscription should not be null");
+ if (subscription == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "Subscription should not be null"));
+ }
return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.UNSUBSCRIBE)
.thenCompose(__ -> pulsar().getNamespaceService().getNamespaceBundleFactory()
@@ -1693,8 +1660,12 @@ public abstract class NamespacesBase extends AdminResource {
@SuppressWarnings("deprecation")
protected CompletableFuture<Void> internalUnsubscribeNamespaceBundleAsync(String subscription, String bundleRange,
boolean authoritative) {
- checkNotNull(subscription, "Subscription should not be null");
- checkNotNull(bundleRange, "BundleRange should not be null");
+ if (subscription == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "Subscription should not be null"));
+ }
+ if (bundleRange == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "BundleRange should not be null"));
+ }
return validateNamespaceOperationAsync(namespaceName, NamespaceOperation.UNSUBSCRIBE)
.thenCompose(__ -> {
@@ -1794,8 +1765,14 @@ public abstract class NamespacesBase extends AdminResource {
}
protected CompletableFuture<Void> internalSetNamespaceAntiAffinityGroupAsync(String antiAffinityGroup) {
- checkNotNull(antiAffinityGroup, "Anti-affinity group should not be null");
- checkNotBlank(antiAffinityGroup, "Anti-affinity group can't be empty");
+ if (antiAffinityGroup == null) {
+ return FutureUtil.failedFuture(
+ new RestException(Status.BAD_REQUEST, "Anti-affinity group should not be null"));
+ }
+ if (StringUtils.isBlank(antiAffinityGroup)) {
+ return FutureUtil.failedFuture(
+ new RestException(Status.PRECONDITION_FAILED, "Anti-affinity group can't be empty"));
+ }
return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE)
.thenCompose(__ -> validatePoliciesReadOnlyAccessAsync()).thenCompose(
__ -> getDefaultBundleDataAsync().thenCompose(
@@ -1833,10 +1810,20 @@ public abstract class NamespacesBase extends AdminResource {
protected CompletableFuture<List<String>> internalGetAntiAffinityNamespacesAsync(String cluster,
String antiAffinityGroup,
String tenant) {
- checkNotNull(cluster, "Cluster should not be null");
- checkNotNull(antiAffinityGroup, "Anti-affinity group should not be null");
- checkNotNull(tenant, "Tenant should not be null");
- checkNotBlank(antiAffinityGroup, "Anti-affinity group can't be empty");
+ if (cluster == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "Cluster should not be null"));
+ }
+ if (antiAffinityGroup == null) {
+ return FutureUtil.failedFuture(
+ new RestException(Status.BAD_REQUEST, "Anti-affinity group should not be null"));
+ }
+ if (tenant == null) {
+ return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, "Tenant should not be null"));
+ }
+ if (StringUtils.isBlank(antiAffinityGroup)) {
+ return FutureUtil.failedFuture(
+ new RestException(Status.PRECONDITION_FAILED, "Anti-affinity group can't be empty"));
+ }
return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ)
.thenCompose(__ -> validateClusterExistsAsync(cluster))
@@ -1867,39 +1854,48 @@ public abstract class NamespacesBase extends AdminResource {
return checkBacklogQuota(quota, retention);
}
- private void clearBacklog(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 != null) {
- if (subscription.startsWith(pulsar().getConfiguration().getReplicatorPrefix())) {
- subscription = PersistentReplicator.getRemoteCluster(subscription);
- }
- for (Topic topic : topicList) {
- if (topic instanceof PersistentTopic
- && !pulsar().getBrokerService().isSystemTopic(TopicName.get(topic.getName()))) {
- futures.add(((PersistentTopic) topic).clearBacklog(subscription));
+ private CompletableFuture<Void> clearBacklogAsync(NamespaceBundle bundle, String subscription) {
+ return pulsar().getNamespaceService().getOwnedPersistentTopicListForNamespaceBundle(bundle)
+ .thenCompose(topicsInBundle -> {
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ String effectiveSubscription = subscription;
+ if (effectiveSubscription != null
+ && effectiveSubscription.startsWith(pulsar().getConfiguration().getReplicatorPrefix())) {
+ effectiveSubscription = PersistentReplicator.getRemoteCluster(effectiveSubscription);
}
- }
- } else {
- for (Topic topic : topicList) {
- if (topic instanceof PersistentTopic
- && !pulsar().getBrokerService().isSystemTopic(TopicName.get(topic.getName()))) {
- futures.add(((PersistentTopic) topic).clearBacklog());
+ final String finalSubscription = effectiveSubscription;
+
+ 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();
+ if (!(loaded instanceof PersistentTopic persistentTopic)) {
+ return CompletableFuture.completedFuture(null);
+ }
+ return finalSubscription != null
+ ? persistentTopic.clearBacklog(finalSubscription)
+ : persistentTopic.clearBacklog();
+ }));
}
- }
- }
- FutureUtil.waitForAll(futures).get();
- } catch (Exception e) {
- log.error("[{}] Failed to clear backlog for namespace {}/{}, subscription: {}", clientAppId(),
- nsName.toString(), bundleRange, subscription, e);
- throw new RestException(e);
- }
+ return FutureUtil.waitForAll(futures);
+ }).exceptionally(ex -> {
+ Throwable cause = FutureUtil.unwrapCompletionException(ex);
+ if (cause instanceof RestException) {
+ throw (RestException) cause;
+ }
+ throw new RestException(cause);
+ });
}
+
private CompletableFuture<Void> unsubscribeAsync(NamespaceBundle bundle, String subscription) {
if (subscription.startsWith(pulsar().getConfiguration().getReplicatorPrefix())) {
return CompletableFuture.failedFuture(
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 1c736d749ee..8e29256c2ca 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
@@ -1375,12 +1375,20 @@ public class Namespaces extends NamespacesBase {
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
try {
validateNamespaceName(property, cluster, namespace);
- internalClearNamespaceBacklog(asyncResponse, authoritative);
} catch (WebApplicationException wae) {
asyncResponse.resume(wae);
+ return;
} catch (Exception e) {
asyncResponse.resume(new RestException(e));
+ return;
}
+ internalClearNamespaceBacklogAsync(authoritative)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to clear backlog on namespace {}", clientAppId(), namespaceName, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@POST
@@ -1390,12 +1398,20 @@ public class Namespaces extends NamespacesBase {
@ApiResponse(code = 307, message = "Current broker doesn't serve the namespace"),
@ApiResponse(code = 403, message = "Don't have admin or operate permission on the namespace"),
@ApiResponse(code = 404, message = "Namespace does not exist") })
- public void clearNamespaceBundleBacklog(@PathParam("property") String property,
+ public void clearNamespaceBundleBacklog(@Suspended final AsyncResponse asyncResponse,
+ @PathParam("property") String property,
@PathParam("cluster") String cluster, @PathParam("namespace") String namespace,
@PathParam("bundle") String bundleRange,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateNamespaceName(property, cluster, namespace);
- internalClearNamespaceBundleBacklog(bundleRange, authoritative);
+ internalClearNamespaceBundleBacklogAsync(bundleRange, authoritative)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to clear backlog on namespace bundle {}/{}", clientAppId(),
+ namespaceName, bundleRange, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@POST
@@ -1410,12 +1426,21 @@ public class Namespaces extends NamespacesBase {
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
try {
validateNamespaceName(property, cluster, namespace);
- internalClearNamespaceBacklogForSubscription(asyncResponse, subscription, authoritative);
} catch (WebApplicationException wae) {
asyncResponse.resume(wae);
+ return;
} catch (Exception e) {
asyncResponse.resume(new RestException(e));
+ return;
}
+ internalClearNamespaceBacklogForSubscriptionAsync(subscription, authoritative)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to clear backlog for subscription {} on namespace {}", clientAppId(),
+ subscription, namespaceName, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@POST
@@ -1425,12 +1450,20 @@ public class Namespaces extends NamespacesBase {
@ApiResponse(code = 307, message = "Current broker doesn't serve the namespace"),
@ApiResponse(code = 403, message = "Don't have admin or operate permission on the namespace"),
@ApiResponse(code = 404, message = "Namespace does not exist") })
- public void clearNamespaceBundleBacklogForSubscription(@PathParam("property") String property,
+ public void clearNamespaceBundleBacklogForSubscription(@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);
- internalClearNamespaceBundleBacklogForSubscription(subscription, bundleRange, authoritative);
+ internalClearNamespaceBundleBacklogForSubscriptionAsync(subscription, bundleRange, authoritative)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to clear backlog for subscription {} 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 836d2699e18..8e6f4fb62a7 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
@@ -1512,14 +1512,14 @@ public class Namespaces extends NamespacesBase {
public void clearNamespaceBacklog(@Suspended final AsyncResponse asyncResponse, @PathParam("tenant") String tenant,
@PathParam("namespace") String namespace,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
- try {
- validateNamespaceName(tenant, namespace);
- internalClearNamespaceBacklog(asyncResponse, authoritative);
- } catch (WebApplicationException wae) {
- asyncResponse.resume(wae);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ validateNamespaceName(tenant, namespace);
+ internalClearNamespaceBacklogAsync(authoritative)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to clear backlog on namespace {}", clientAppId(), namespaceName, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@POST
@@ -1530,11 +1530,19 @@ public class Namespaces extends NamespacesBase {
@ApiResponse(code = 307, message = "Current broker doesn't serve the namespace"),
@ApiResponse(code = 403, message = "Don't have admin or operate permission on the namespace"),
@ApiResponse(code = 404, message = "Namespace does not exist") })
- public void clearNamespaceBundleBacklog(@PathParam("tenant") String tenant,
+ public void clearNamespaceBundleBacklog(@Suspended final AsyncResponse asyncResponse,
+ @PathParam("tenant") String tenant,
@PathParam("namespace") String namespace, @PathParam("bundle") String bundleRange,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateNamespaceName(tenant, namespace);
- internalClearNamespaceBundleBacklog(bundleRange, authoritative);
+ internalClearNamespaceBundleBacklogAsync(bundleRange, authoritative)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to clear backlog on namespace bundle {}/{}", clientAppId(),
+ namespaceName, bundleRange, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@POST
@@ -1548,14 +1556,15 @@ public class Namespaces extends NamespacesBase {
@PathParam("tenant") String tenant, @PathParam("namespace") String namespace,
@PathParam("subscription") String subscription,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
- try {
- validateNamespaceName(tenant, namespace);
- internalClearNamespaceBacklogForSubscription(asyncResponse, subscription, authoritative);
- } catch (WebApplicationException wae) {
- asyncResponse.resume(wae);
- } catch (Exception e) {
- asyncResponse.resume(new RestException(e));
- }
+ validateNamespaceName(tenant, namespace);
+ internalClearNamespaceBacklogForSubscriptionAsync(subscription, authoritative)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to clear backlog for subscription {} on namespace {}", clientAppId(),
+ subscription, namespaceName, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, ex);
+ return null;
+ });
}
@POST
@@ -1566,12 +1575,20 @@ public class Namespaces extends NamespacesBase {
@ApiResponse(code = 307, message = "Current broker doesn't serve the namespace"),
@ApiResponse(code = 403, message = "Don't have admin or operate permission on the namespace"),
@ApiResponse(code = 404, message = "Namespace does not exist") })
- public void clearNamespaceBundleBacklogForSubscription(@PathParam("tenant") String tenant,
+ public void clearNamespaceBundleBacklogForSubscription(@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);
- internalClearNamespaceBundleBacklogForSubscription(subscription, bundleRange, authoritative);
+ internalClearNamespaceBundleBacklogForSubscriptionAsync(subscription, bundleRange, authoritative)
+ .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+ .exceptionally(ex -> {
+ log.error("[{}] Failed to clear backlog for subscription {} 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/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index f17526416e5..2a40cfa4464 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -1388,6 +1388,25 @@ public class NamespaceService implements AutoCloseable {
});
}
+ public CompletableFuture<List<String>> getOwnedPersistentTopicListForNamespaceBundle(NamespaceBundle bundle) {
+ return getListOfPersistentTopics(bundle.getNamespaceObject()).thenCompose(topics ->
+ CompletableFuture.completedFuture(
+ topics.stream()
+ .filter(topic -> bundle.includes(TopicName.get(topic)))
+ .collect(Collectors.toList())))
+ .thenCombine(getPartitions(bundle.getNamespaceObject(), TopicDomain.persistent).thenCompose(topics ->
+ CompletableFuture.completedFuture(
+ topics.stream().filter(topic -> bundle.includes(TopicName.get(topic)))
+ .collect(Collectors.toList()))), (left, right) -> {
+ for (String topic : right) {
+ if (!left.contains(topic)) {
+ left.add(topic);
+ }
+ }
+ return left;
+ });
+ }
+
/***
* Checks whether the topic exists( partitioned or non-partitioned ).
*/
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 2c5d045f758..9a5c72ff240 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
@@ -2284,6 +2284,132 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
assertEquals(backlog, 0);
}
+ @Test(dataProvider = "numBundles")
+ public void testClearNamespaceBundleBacklogOnUnloadedBundle(Integer numBundles) throws Exception {
+ String namespace = "prop-xyz/ns1-bundles";
+ admin.namespaces().createNamespace("prop-xyz/ns1-bundles", numBundles);
+
+ String topic = "persistent://" + namespace + "/t1";
+ String subscription = "sub1";
+
+ Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionName(subscription)
+ .subscribe();
+
+ Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
+ .topic(topic)
+ .enableBatching(false)
+ .messageRoutingMode(MessageRoutingMode.SinglePartition)
+ .create();
+ for (int i = 0; i < 10; i++) {
+ producer.send(("message-" + i).getBytes());
+ }
+ producer.close();
+ consumer.close();
+
+ long backlog = admin.topics().getStats(topic).getSubscriptions().get(subscription).getMsgBacklog();
+ assertEquals(backlog, 10);
+
+ 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().clearNamespaceBundleBacklog(namespace, bundle.getBundleRange());
+
+ @Cleanup
+ Consumer<byte[]> consumer1 = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionName(subscription)
+ .subscribe();
+
+ long backlogAfter = admin.topics().getStats(topic).getSubscriptions().get(subscription).getMsgBacklog();
+ assertEquals(backlogAfter, 0);
+ }
+
+ @Test(dataProvider = "numBundles")
+ public void testClearNamespaceBundleBacklogForSubscriptionOnUnloadedBundle(Integer numBundles) throws Exception {
+ String namespace = "prop-xyz/ns1-bundles";
+ admin.namespaces().createNamespace("prop-xyz/ns1-bundles", 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();
+
+ Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
+ .topic(topic)
+ .enableBatching(false)
+ .messageRoutingMode(MessageRoutingMode.SinglePartition)
+ .create();
+ for (int i = 0; i < 10; i++) {
+ producer.send(("message-" + i).getBytes());
+ }
+ producer.close();
+ consumer1.close();
+ consumer2.close();
+
+ long backlog = admin.topics().getStats(topic).getSubscriptions().get(subscription).getMsgBacklog();
+ assertEquals(backlog, 10);
+ long otherBacklog = admin.topics().getStats(topic).getSubscriptions().get(otherSubscription).getMsgBacklog();
+ assertEquals(otherBacklog, 10);
+
+ 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().clearNamespaceBundleBacklogForSubscription(namespace, bundle.getBundleRange(),
+ subscription);
+
+ @Cleanup
+ Consumer<byte[]> consumer3 = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionName(subscription)
+ .subscribe();
+
+ @Cleanup
+ Consumer<byte[]> consumer4 = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionName(otherSubscription)
+ .subscribe();
+
+ long backlogAfter = admin.topics().getStats(topic).getSubscriptions().get(subscription).getMsgBacklog();
+ assertEquals(backlogAfter, 0);
+ long otherBacklogAfter =
+ admin.topics().getStats(topic).getSubscriptions().get(otherSubscription).getMsgBacklog();
+ assertEquals(otherBacklogAfter, 10);
+ }
+
+
@Test(dataProvider = "bundling")
public void testUnsubscribeOnNamespace(Integer numBundles) throws Exception {
admin.namespaces().createNamespace("prop-xyz/ns1-bundles", numBundles);
(pulsar) 02/11: Revert "[fix][admin] Refactor namespace migration operation to async in rest api (#25478)"
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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit eefa9f0fc049b8ff225c3fa56e77434f44217cbf
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Wed Apr 15 17:34:20 2026 +0300
Revert "[fix][admin] Refactor namespace migration operation to async in rest api (#25478)"
This reverts commit 066a3f30a69c1e6c33902ed4716bc4d8d6bcc2e9.
---
.../pulsar/broker/admin/impl/NamespacesBase.java | 38 ++++++--
.../apache/pulsar/broker/admin/v2/Namespaces.java | 12 +--
.../pulsar/broker/admin/NamespacesV2Test.java | 102 +--------------------
3 files changed, 33 insertions(+), 119 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 3b22ea64571..8e1d11b15f3 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
@@ -2847,15 +2847,22 @@ public abstract class NamespacesBase extends AdminResource {
});
}
- protected CompletableFuture<Void> internalEnableMigrationAsync(boolean migrated) {
- return validateSuperUserAccessAsync().thenCompose(__ -> getDefaultBundleDataAsync().thenCompose(
- defaultBundleData -> getLocalPolicies().setLocalPoliciesWithCreateAsync(namespaceName,
- oldPolicies -> oldPolicies.map(
- policies -> new LocalPolicies(policies.bundles,
- policies.bookieAffinityGroup,
- policies.namespaceAntiAffinityGroup, migrated))
- .orElseGet(() -> new LocalPolicies(defaultBundleData, null, null, migrated)))))
- .thenAccept(__ -> log.info("Successfully updated migration on namespace {}", namespaceName));
+ protected void internalEnableMigration(boolean migrated) {
+ validateSuperUserAccess();
+ try {
+ getLocalPolicies().setLocalPoliciesWithCreate(namespaceName, oldPolicies -> oldPolicies.map(
+ policies -> new LocalPolicies(policies.bundles,
+ policies.bookieAffinityGroup,
+ policies.namespaceAntiAffinityGroup,
+ migrated))
+ .orElseGet(() -> new LocalPolicies(getDefaultBundleData(), null, null, migrated)));
+ log.info("Successfully updated migration on namespace {}", namespaceName);
+ } catch (RestException re) {
+ throw re;
+ } catch (Exception e) {
+ log.error("Failed to update migration on namespace {}", namespaceName, e);
+ throw new RestException(e);
+ }
}
protected Policies getDefaultPolicesIfNull(Policies policies) {
@@ -2953,6 +2960,19 @@ public abstract class NamespacesBase extends AdminResource {
.thenApply(policies -> policies.allowed_clusters);
}
+ // TODO remove this sync method after async refactor
+ @Deprecated
+ private BundlesData getDefaultBundleData() {
+ try {
+ return getDefaultBundleDataAsync().get(config().getMetadataStoreOperationTimeoutSeconds(),
+ TimeUnit.SECONDS);
+ } catch (Exception e) {
+ log.error("[{}] Failed to get namespace-policy configuration for namespace {}", clientAppId(),
+ namespaceName, e);
+ throw new RestException(e);
+ }
+ }
+
private CompletableFuture<BundlesData> getDefaultBundleDataAsync() {
return namespaceResources().getPoliciesAsync(namespaceName).thenApply(
optionalPolicies -> optionalPolicies.isPresent() ? optionalPolicies.get().bundles :
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 29f652b10f6..a997e1d898a 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
@@ -3066,19 +3066,11 @@ public class Namespaces extends NamespacesBase {
@ApiResponse(code = 200, message = "Operation successful"),
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Property or cluster or namespace doesn't exist") })
- public void enableMigration(@Suspended AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
+ public void enableMigration(@PathParam("tenant") String tenant,
@PathParam("namespace") String namespace,
boolean migrated) {
validateNamespaceName(tenant, namespace);
- internalEnableMigrationAsync(migrated)
- .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
- .exceptionally(ex -> {
- log.error("[{}] Failed to update migration, tenant: {}, namespace: {}",
- clientAppId(), tenant, namespace, ex);
- resumeAsyncResponseExceptionally(asyncResponse, ex);
- return null;
- });
+ internalEnableMigration(migrated);
}
@POST
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java
index 9fd1ea1dc01..629e92c056f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java
@@ -24,7 +24,6 @@ import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
@@ -353,8 +352,7 @@ public class NamespacesV2Test extends MockedPulsarServiceBaseTest {
// 2.set enable migration
boolean enableMigrationReq = true;
- asyncRequests(response -> namespaces.enableMigration(response, testTenant, enableMigrationGroupNs,
- enableMigrationReq));
+ namespaces.enableMigration(testTenant, enableMigrationGroupNs, enableMigrationReq);
// 3.query namespace num bundles, should be conf.getDefaultNumberOfNamespaceBundles()
BundlesData bundlesData = (BundlesData) asyncRequests(
@@ -378,8 +376,7 @@ public class NamespacesV2Test extends MockedPulsarServiceBaseTest {
// 2.set enable migration
boolean enableMigrationReq = true;
- asyncRequests(response -> namespaces.enableMigration(response, testTenant, enableMigrationGroupNs,
- enableMigrationReq));
+ namespaces.enableMigration(testTenant, enableMigrationGroupNs, enableMigrationReq);
// 3.query namespace num bundles, should be policies.bundles, which we set before
BundlesData bundlesData = (BundlesData) asyncRequests(
@@ -421,99 +418,4 @@ public class NamespacesV2Test extends MockedPulsarServiceBaseTest {
assertNull(bookieAffinityGroupDataResp);
}
- @Test
- public void testSetAndDeleteNamespaceAntiAffinityGroup() throws Exception {
- // 1. create namespace with empty policies, namespace anti affinity group should be null
- String setNamespaceAntiAffinityGroupNs = "test-set-namespace-anti-affinity-group-ns";
- asyncRequests(
- response -> namespaces.createNamespace(response, testTenant, setNamespaceAntiAffinityGroupNs, null));
- String namespaceAntiAffinityGroupResp = (String) asyncRequests(
- response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant,
- setNamespaceAntiAffinityGroupNs));
- assertNull(namespaceAntiAffinityGroupResp);
-
- // 2.set namespace anti affinity group
- String namespaceAntiAffinityGroupReq = "namespace-anti-affinity-group";
- asyncRequests(response -> namespaces.setNamespaceAntiAffinityGroup(response, testTenant,
- setNamespaceAntiAffinityGroupNs, namespaceAntiAffinityGroupReq));
-
- // 3.assert namespace anti affinity group
- namespaceAntiAffinityGroupResp = (String) asyncRequests(
- response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant,
- setNamespaceAntiAffinityGroupNs));
- assertEquals(namespaceAntiAffinityGroupResp, namespaceAntiAffinityGroupReq);
-
- // 4.delete namespace anti affinity group
- asyncRequests(response -> namespaces.removeNamespaceAntiAffinityGroup(response, testTenant,
- setNamespaceAntiAffinityGroupNs));
-
- // 5.assert namespace anti affinity group
- namespaceAntiAffinityGroupResp = (String) asyncRequests(
- response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant,
- setNamespaceAntiAffinityGroupNs));
- assertNull(namespaceAntiAffinityGroupResp);
- }
-
- @Test
- @SuppressWarnings("unchecked")
- public void testGetClusterAntiAffinityNamespaces() throws Exception {
- // create 5 namespaces, 3 namespaces are set to the same namespace anti affinity group,
- // 2 namespaces are not set to any anti affinity group
- String namespaceWithAntiAffinity1 = "namespace-with-anti-affinity-1";
- String namespaceWithAntiAffinity2 = "namespace-with-anti-affinity-2";
- String namespaceWithAntiAffinity3 = "namespace-with-anti-affinity-3";
- String namespaceWithoutAntiAffinity1 = "namespace-without-anti-affinity-1";
- String namespaceWithoutAntiAffinity2 = "namespace-without-anti-affinity-2";
-
- // create namespaces
- List<String> allNamespaces =
- List.of(namespaceWithAntiAffinity1, namespaceWithAntiAffinity2, namespaceWithAntiAffinity3,
- namespaceWithoutAntiAffinity1, namespaceWithoutAntiAffinity2);
- for (String namespace : allNamespaces) {
- asyncRequests(response -> namespaces.createNamespace(response, testTenant, namespace, null));
- }
-
- // set namespace anti affinity group
- String namespaceAntiAffinityGroupReq = "namespace-anti-affinity-group";
- List<String> namespacesWithAntiAffinityGroup =
- List.of(namespaceWithAntiAffinity1, namespaceWithAntiAffinity2, namespaceWithAntiAffinity3);
- for (String namespace : namespacesWithAntiAffinityGroup) {
- asyncRequests(response -> namespaces.setNamespaceAntiAffinityGroup(response, testTenant, namespace,
- namespaceAntiAffinityGroupReq));
- }
-
- // assert namespace anti affinity group
- for (String namespace : namespacesWithAntiAffinityGroup) {
- String namespaceAntiAffinityGroupResp = (String) asyncRequests(
- response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant, namespace));
- assertEquals(namespaceAntiAffinityGroupResp, namespaceAntiAffinityGroupReq);
- }
-
- // get namespaces in cluster of given anti affinity group
- List<String> namespacesResp = (List<String>) asyncRequests(
- response -> namespaces.getAntiAffinityNamespaces(response, testLocalCluster,
- namespaceAntiAffinityGroupReq, testTenant));
- List<String> namespacesWithFullPath =
- namespacesWithAntiAffinityGroup.stream().map(ns -> NamespaceName.get(testTenant, ns))
- .map(NamespaceName::toString).toList();
- assertEquals(namespacesResp, namespacesWithFullPath);
- }
-
- @Test
- public void testEnableMigrationAndDisableMigration() throws Exception {
- String enableMigrationGroupNs = "test-enable-migration-disable-migration-ns";
- asyncRequests(response -> namespaces.createNamespace(response, testTenant, enableMigrationGroupNs, null));
-
- // Enable migration
- asyncRequests(response -> namespaces.enableMigration(response, testTenant, enableMigrationGroupNs, true));
- Policies policiesResp = (Policies) asyncRequests(
- response -> namespaces.getPolicies(response, testTenant, enableMigrationGroupNs));
- assertTrue(policiesResp.migrated);
-
- // Disable migration
- asyncRequests(response -> namespaces.enableMigration(response, testTenant, enableMigrationGroupNs, false));
- policiesResp = (Policies) asyncRequests(
- response -> namespaces.getPolicies(response, testTenant, enableMigrationGroupNs));
- assertFalse(policiesResp.migrated);
- }
}
(pulsar) 05/11: [fix][client] Fix async APIs to return failed futures on validation errors (#25287)
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.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9a969707abd0cfebd32118ca85ae69fddbcba8c4
Author: Hao Zhang <zh...@cmss.chinamobile.com>
AuthorDate: Mon Mar 9 19:42:23 2026 +0800
[fix][client] Fix async APIs to return failed futures on validation errors (#25287)
Co-authored-by: 张浩 <zh...@100.me>
(cherry picked from commit de5221b2a45f17e2de17ec5bdf1ab96c3e4b309e)
---
.../pulsar/client/impl/MessageChunkingTest.java | 37 ----------------------
.../apache/pulsar/client/impl/ConsumerBase.java | 11 +++++--
.../client/impl/MultiTopicsConsumerImpl.java | 4 ++-
.../pulsar/client/impl/ProducerBuilderImpl.java | 6 ++--
.../client/impl/MultiTopicsConsumerImplTest.java | 11 +++++++
.../client/impl/ProducerBuilderImplTest.java | 20 ++++++++++++
6 files changed, 46 insertions(+), 43 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java
index 1a122b94416..d8a3698636d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java
@@ -99,19 +99,6 @@ public class MessageChunkingTest extends ProducerConsumerBase {
return new Object[][] { { true }, { false } };
}
- @Test
- public void testInvalidConfig() throws Exception {
- final String topicName = "persistent://my-property/my-ns/my-topic1";
- ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer().topic(topicName);
- // batching and chunking can't be enabled together
- try {
- Producer<byte[]> producer = producerBuilder.enableChunking(true).enableBatching(true).create();
- fail("producer creation should have fail");
- } catch (IllegalArgumentException ie) {
- // Ok
- }
- }
-
@Test(dataProvider = "ackReceiptEnabledWithMaxMessageSize")
public void testLargeMessage(boolean ackReceiptEnabled, boolean clientSizeMaxMessageSize) throws Exception {
@@ -448,30 +435,6 @@ public class MessageChunkingTest extends ProducerConsumerBase {
Assert.assertEquals(((ConsumerImpl<String>) consumer).getAvailablePermits(), 8);
}
- /**
- * Validate that chunking is not supported with batching and non-persistent topic.
- *
- * @throws Exception
- */
- @Test
- public void testInvalidUseCaseForChunking() throws Exception {
-
- log.info("-- Starting {} test --", methodName);
- this.conf.setMaxMessageSize(5);
- final String topicName = "persistent://my-property/my-ns/my-topic1";
-
- ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer().topic(topicName);
-
- try {
- Producer<byte[]> producer = producerBuilder.enableChunking(true).enableBatching(true).create();
- fail("it should have failied because chunking can't be used with batching enabled");
- } catch (IllegalArgumentException ie) {
- // Ok
- }
-
- log.info("-- Exiting {} test --", methodName);
- }
-
@Test
public void testExpireIncompleteChunkMessage() throws Exception{
final String topicName = "persistent://prop/use/ns-abc/expireMsg";
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
index 1dfbf51f507..4e0f4466925 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
@@ -18,7 +18,6 @@
*/
package org.apache.pulsar.client.impl;
-import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.pulsar.common.protocol.Commands.DEFAULT_CONSUMER_EPOCH;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Queues;
@@ -657,7 +656,10 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
}
TransactionImpl txnImpl = null;
if (null != txn) {
- checkArgument(txn instanceof TransactionImpl);
+ if (!(txn instanceof TransactionImpl)) {
+ return FutureUtil.failedFuture(new IllegalArgumentException(
+ "Expected txn to be an instance of TransactionImpl, but got " + txn.getClass().getName()));
+ }
txnImpl = (TransactionImpl) txn;
CompletableFuture<Void> completableFuture = new CompletableFuture<>();
if (!txnImpl.checkIfOpen(completableFuture)) {
@@ -686,7 +688,10 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
TransactionImpl txnImpl = null;
if (null != txn) {
- checkArgument(txn instanceof TransactionImpl);
+ if (!(txn instanceof TransactionImpl)) {
+ return FutureUtil.failedFuture(new IllegalArgumentException(
+ "Expected txn to be an instance of TransactionImpl, but got " + txn.getClass().getName()));
+ }
txnImpl = (TransactionImpl) txn;
}
return doAcknowledgeWithTxn(messageId, AckType.Cumulative, Collections.emptyMap(), txnImpl);
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
index 014b883373b..f5f24cb5b89 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
@@ -1266,7 +1266,9 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
// un-subscribe a given topic
public CompletableFuture<Void> unsubscribeAsync(String topicName) {
- checkArgument(TopicName.isValid(topicName), "Invalid topic name:" + topicName);
+ if (!TopicName.isValid(topicName)) {
+ return FutureUtil.failedFuture(new IllegalArgumentException("Invalid topic name:" + topicName));
+ }
if (getState() == State.Closing || getState() == State.Closed) {
return FutureUtil.failedFuture(
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java
index 0b5399ee258..3457aac0402 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java
@@ -93,8 +93,10 @@ public class ProducerBuilderImpl<T> implements ProducerBuilder<T> {
@Override
public CompletableFuture<Producer<T>> createAsync() {
// config validation
- checkArgument(!(conf.isBatchingEnabled() && conf.isChunkingEnabled()),
- "Batching and chunking of messages can't be enabled together");
+ if (conf.isBatchingEnabled() && conf.isChunkingEnabled()) {
+ return FutureUtil.failedFuture(
+ new IllegalArgumentException("Batching and chunking of messages can't be enabled together"));
+ }
if (conf.getTopicName() == null) {
return FutureUtil
.failedFuture(new IllegalArgumentException("Topic name must be set on the producer builder"));
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImplTest.java
index 02a4d2ebba8..54175613e3b 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImplTest.java
@@ -39,6 +39,7 @@ import io.netty.util.concurrent.DefaultThreadFactory;
import java.util.Arrays;
import java.util.HashSet;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -223,6 +224,16 @@ public class MultiTopicsConsumerImplTest {
verify(clientMock, times(1)).cleanupConsumer(any());
}
+ @Test
+ public void testUnsubscribeAsyncInvalidTopicNameReturnsFailedFuture() {
+ MultiTopicsConsumerImpl<byte[]> consumer = createMultiTopicsConsumer();
+ CompletableFuture<Void> future = consumer.unsubscribeAsync("persistent://public/invalid-topic");
+
+ assertTrue(future.isCompletedExceptionally());
+ CompletionException ex = expectThrows(CompletionException.class, future::join);
+ assertTrue(ex.getCause() instanceof IllegalArgumentException);
+ }
+
@Test
public void testDontCheckForPartitionsUpdatesOnNonPartitionedTopics() throws Exception {
ExecutorProvider executorProvider = mock(ExecutorProvider.class);
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java
index bd96cf27aee..a49114d67b5 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java
@@ -23,9 +23,12 @@ import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageRouter;
@@ -383,6 +386,23 @@ public class ProducerBuilderImplTest {
producerBuilderImpl.maxPendingMessagesAcrossPartitions(1000);
}
+ @Test
+ public void testCreateAsyncFailsWhenBatchingAndChunkingEnabled() {
+ producerBuilderImpl = new ProducerBuilderImpl<>(client, Schema.BYTES);
+ CompletableFuture<Producer<byte[]>> future = producerBuilderImpl.topic(TOPIC_NAME)
+ .enableBatching(true)
+ .enableChunking(true)
+ .createAsync();
+
+ assertTrue(future.isCompletedExceptionally());
+ try {
+ future.join();
+ fail("Expected IllegalArgumentException");
+ } catch (CompletionException e) {
+ assertTrue(e.getCause() instanceof IllegalArgumentException);
+ }
+ }
+
private class CustomMessageRouter implements MessageRouter {
@Override
public int choosePartition(Message<?> msg, TopicMetadata metadata) {