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/08/10 09:02:29 UTC
(pulsar) 01/02: [fix][admin] Avoid creating subscriptions when peeking messages if auto-creation is disabled (#26279)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 76c3e2ea6aa8930518d28de63b4d40c050bc5815
Author: void-ptr974 <vo...@protonmail.com>
AuthorDate: Fri Aug 7 12:32:13 2026 +0800
[fix][admin] Avoid creating subscriptions when peeking messages if auto-creation is disabled (#26279)
(cherry picked from commit 2fbf416e39cb506fbe0b77fc1a92dd5facb80293)
---
.../broker/admin/impl/PersistentTopicsBase.java | 44 ++++++++++++++--------
.../pulsar/broker/admin/v2/PersistentTopics.java | 4 +-
.../pulsar/broker/admin/PersistentTopicsTest.java | 12 ++++++
.../org/apache/pulsar/client/admin/Topics.java | 4 ++
4 files changed, 48 insertions(+), 16 deletions(-)
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 d95434b57179..9513177237aa 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
@@ -2839,9 +2839,8 @@ public class PersistentTopicsBase extends AdminResource {
PersistentReplicator repl = getReplicatorReference(subName, (PersistentTopic) topic);
entry = repl.peekNthMessage(messagePosition);
} else {
- PersistentSubscription sub =
- (PersistentSubscription) getSubscriptionReference(subName, (PersistentTopic) topic);
- entry = sub.peekNthMessage(messagePosition);
+ entry = findOrCreateSubscriptionAsync(subName, (PersistentTopic) topic)
+ .thenCompose(sub -> sub.peekNthMessage(messagePosition));
}
}
return entry.thenApply(e -> Pair.of(e, (PersistentTopic) topic));
@@ -4347,20 +4346,35 @@ public class PersistentTopicsBase extends AdminResource {
}
/**
- * Get the Subscription object reference from the Topic reference.
+ * Find the subscription or create it when automatic subscription creation is allowed.
*/
- private Subscription getSubscriptionReference(String subName, PersistentTopic topic) {
- try {
- Subscription sub = topic.getSubscription(subName);
- if (sub == null) {
- sub = topic.createSubscription(subName,
- InitialPosition.Earliest, false, null).get();
- }
-
- return checkNotNull(sub);
- } catch (Exception e) {
- throw new RestException(Status.NOT_FOUND, getSubNotFoundErrorMessage(topicName.toString(), subName));
+ private CompletableFuture<Subscription> findOrCreateSubscriptionAsync(String subName, PersistentTopic topic) {
+ Subscription subscription = topic.getSubscription(subName);
+ if (subscription != null) {
+ return CompletableFuture.completedFuture(subscription);
}
+
+ return pulsar().getBrokerService().isAllowAutoSubscriptionCreationAsync(topicName)
+ .thenCompose(isAllowed -> {
+ Subscription existingSubscription = topic.getSubscription(subName);
+ if (existingSubscription != null) {
+ return CompletableFuture.completedFuture(existingSubscription);
+ }
+ if (!isAllowed) {
+ return CompletableFuture.failedFuture(new RestException(Status.PRECONDITION_FAILED,
+ String.format("Subscription %s does not exist for topic %s and automatic "
+ + "subscription creation is disabled",
+ subName, topicName)));
+ }
+ return topic.createSubscription(subName, InitialPosition.Earliest, false, null)
+ .handle((createdSubscription, ex) -> {
+ if (ex != null || createdSubscription == null) {
+ throw new RestException(Status.NOT_FOUND,
+ getSubNotFoundErrorMessage(topicName.toString(), subName));
+ }
+ return createdSubscription;
+ });
+ });
}
/**
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
index 3d54d73ff45b..ee06b73ac9ba 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
@@ -2011,7 +2011,9 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiResponse(code = 404, message = "Namespace or topic, subscription or the message position does not"
+ " exist"),
@ApiResponse(code = 405, message = "Skipping messages on a non-persistent topic is not allowed"),
- @ApiResponse(code = 412, message = "Topic name is not valid"),
+ @ApiResponse(code = 412,
+ message = "Topic name is not valid, or the subscription does not exist and automatic "
+ + "subscription creation is disabled"),
@ApiResponse(code = 500, message = "Internal server error"),
@ApiResponse(code = 503, message = "Failed to validate global cluster configuration")})
public void peekNthMessage(
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
index 4cccebff0f5e..1e7c0f456cb5 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java
@@ -100,6 +100,7 @@ import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.policies.data.AuthAction;
+import org.apache.pulsar.common.policies.data.AutoSubscriptionCreationOverride;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
import org.apache.pulsar.common.policies.data.Policies;
@@ -1146,9 +1147,20 @@ public class PersistentTopicsTest extends MockedPulsarServiceBaseTest {
producer.send("test" + i);
}
+ admin.namespaces().setAutoSubscriptionCreation("tenant-xyz/ns-abc",
+ AutoSubscriptionCreationOverride.builder().allowAutoSubscriptionCreation(false).build());
+ PulsarAdminException.PreconditionFailedException exception = Assert.expectThrows(
+ PulsarAdminException.PreconditionFailedException.class,
+ () -> admin.topics().peekMessages(partitionedTopic, subscriptionName, 3));
+ Assert.assertEquals(exception.getStatusCode(), Response.Status.PRECONDITION_FAILED.getStatusCode());
+ Assert.assertTrue(admin.topics().getSubscriptions(partitionedTopic).isEmpty());
+
+ admin.namespaces().setAutoSubscriptionCreation("tenant-xyz/ns-abc",
+ AutoSubscriptionCreationOverride.builder().allowAutoSubscriptionCreation(true).build());
List<Message<byte[]>> messages = admin.topics().peekMessages(partitionedTopic, subscriptionName, 3);
Assert.assertEquals(messages.size(), 3);
+ Assert.assertEquals(admin.topics().getSubscriptions(partitionedTopic), List.of(subscriptionName));
producer.close();
}
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
index e68be8fd2e80..4737f7426101 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
@@ -1500,6 +1500,8 @@ public interface Topics {
* Don't have admin permission
* @throws NotFoundException
* Topic or subscription does not exist
+ * @throws PreconditionFailedException
+ * Subscription does not exist and automatic subscription creation is disabled
* @throws PulsarAdminException
* Unexpected error
*/
@@ -1533,6 +1535,8 @@ public interface Topics {
* Don't have admin permission
* @throws NotFoundException
* Topic or subscription does not exist
+ * @throws PreconditionFailedException
+ * Subscription does not exist and automatic subscription creation is disabled
* @throws PulsarAdminException
* Unexpected error
*/