You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2026/08/20 08:56:04 UTC
(pulsar) branch master updated: [fix][broker] Reject topic names with leading or trailing whitespace on creation (#26281)
This is an automated email from the ASF dual-hosted git repository.
nodece pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 20ef856da92b [fix][broker] Reject topic names with leading or trailing whitespace on creation (#26281)
20ef856da92b is described below
commit 20ef856da92b2b69bca462ab625e37ee0c0af7a5
Author: crossoverJie <cr...@gmail.com>
AuthorDate: Thu Aug 20 16:55:57 2026 +0800
[fix][broker] Reject topic names with leading or trailing whitespace on creation (#26281)
Signed-off-by: crossoverJie <cr...@gmail.com>
---
.../apache/pulsar/broker/admin/AdminResource.java | 24 +++++++++
.../broker/admin/impl/PersistentTopicsBase.java | 7 ---
.../broker/admin/v2/NonPersistentTopics.java | 1 +
.../pulsar/broker/admin/v2/ScalableTopics.java | 5 +-
.../pulsar/broker/service/BrokerService.java | 11 ++++
.../pulsar/broker/admin/PersistentTopicsTest.java | 44 ++++++++++++++++
.../admin/ScalableTopicsListByPropertyTest.java | 59 ++++++++++++++++++++++
.../BrokerServiceAutoTopicCreationTest.java | 25 +++++++++
.../pulsar/client/api/ConsumerCreationTest.java | 36 +++++++++++++
.../admin/internal/NonPersistentTopicsImpl.java | 9 ++++
.../client/admin/internal/ScalableTopicsImpl.java | 9 ++++
.../pulsar/client/admin/internal/TopicsImpl.java | 14 +++++
.../org/apache/pulsar/common/naming/TopicName.java | 31 ++++++++++++
.../apache/pulsar/common/naming/TopicNameTest.java | 35 +++++++++++++
14 files changed, 302 insertions(+), 8 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 b8214962324f..182b25dbeb13 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
@@ -56,6 +56,7 @@ import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.admin.internal.TopicsImpl;
import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.naming.SystemTopicNames;
import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
@@ -247,6 +248,29 @@ public abstract class AdminResource extends PulsarWebResource {
}
}
+ /**
+ * Validates that a topic can be created.
+ *
+ * <p>This is the single source of truth for topic-creation name validation shared by every admin create
+ * endpoint (persistent, non-persistent and scalable topics). Rejecting here keeps topics which could never be
+ * reached (e.g. because clients trim topic names) from being created. The transaction-internal-name rule is
+ * gated on {@link TopicDomain#persistent} so it stays specific to persistent topics, while the whitespace
+ * validation applies uniformly to all topic types.
+ */
+ protected void validateCreateTopic(TopicName topicName) {
+ if (topicName.getDomain() == TopicDomain.persistent
+ && SystemTopicNames.isTransactionInternalName(topicName)) {
+ log.warn().attr("topic", topicName).log("Forbidden to create transaction internal topic");
+ throw new RestException(Status.BAD_REQUEST, "Cannot create topic in system topic format!");
+ }
+ try {
+ TopicName.validateTopicNameForCreation(topicName);
+ } catch (IllegalArgumentException e) {
+ log.warn().attr("topic", topicName).log("Forbidden to create topic with an invalid name");
+ throw new RestException(Status.PRECONDITION_FAILED, e.getMessage());
+ }
+ }
+
protected void validatePersistentTopicName(String tenant, String namespace, String encodedTopic) {
validateTopicName(tenant, namespace, encodedTopic);
if (topicName.getDomain() != TopicDomain.persistent) {
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 2b39835dbf23..f0215fb69f7e 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
@@ -224,13 +224,6 @@ public class PersistentTopicsBase extends AdminResource {
.thenCompose(__ -> getAuthorizationService().getPermissionsAsync(topicName));
}
- protected void validateCreateTopic(TopicName topicName) {
- if (isTransactionInternalName(topicName)) {
- log.warn().attr("topic", topicName).log("Forbidden to create transaction internal topic");
- throw new RestException(Status.BAD_REQUEST, "Cannot create topic in system topic format!");
- }
- }
-
public void validateAdminOperationOnTopic(boolean authoritative) {
validateAdminAccessForTenant(topicName.getTenant());
validateTopicOwnership(topicName, authoritative);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java
index cec46d6ce452..6a0493655dee 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java
@@ -220,6 +220,7 @@ public class NonPersistentTopics extends PersistentTopics {
validateNamespaceName(tenant, namespace);
validateGlobalNamespaceOwnership();
validateTopicName(tenant, namespace, encodedTopic);
+ validateCreateTopic(topicName);
internalCreatePartitionedTopic(asyncResponse, numPartitions, createLocalTopicOnly, properties);
} catch (Exception e) {
log.error()
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/ScalableTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/ScalableTopics.java
index ff9b16ccf204..ce253f6fdfb9 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/ScalableTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/ScalableTopics.java
@@ -71,6 +71,7 @@ import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.common.scalable.ScalableTopicConstants;
import org.apache.pulsar.common.scalable.SegmentInfo;
import org.apache.pulsar.common.scalable.SegmentTopicName;
+import org.apache.pulsar.common.util.Codec;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.metadata.api.MetadataStoreException;
@@ -178,7 +179,9 @@ public class ScalableTopics extends AdminResource {
@RequestBody(description = "Key value pair properties for the topic metadata")
Map<String, String> properties) {
validateNamespaceName(tenant, namespace);
- TopicName tn = TopicName.get(TopicDomain.topic.value(), namespaceName, encodedTopic);
+ String decodedTopic = Codec.decode(encodedTopic);
+ TopicName tn = TopicName.get(TopicDomain.topic.value(), namespaceName, decodedTopic);
+ validateCreateTopic(tn);
validateNamespaceOperationAsync(namespaceName, NamespaceOperation.CREATE_TOPIC)
.thenCompose(__ -> {
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 0019cf562c3e..406dbb3ec5e4 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
@@ -4213,6 +4213,17 @@ public class BrokerService implements Closeable {
return CompletableFuture.completedFuture(false);
}
+ // A topic whose name has surrounding whitespace could never be used: Pulsar clients trim topic names, so
+ // producing to or consuming from it would target the trimmed name instead. Refuse to auto-create it, which
+ // also covers clients that do not trim the name themselves.
+ // Note that topics which already have such a name are unaffected: they are loaded, not created here.
+ if (!TopicName.isValidForCreation(topicName)) {
+ log.warn()
+ .attr("topic", topicName)
+ .log("Preventing AutoTopicCreation of a topic whose name has surrounding whitespace");
+ return CompletableFuture.completedFuture(false);
+ }
+
// ExtensibleLoadManagerImpl.internal topics expects to be non-partitioned-topics now.
// We don't allow the auto-creation here.
// ExtensibleLoadManagerImpl.start() is responsible to create the internal system topics.
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 9c785e8a2e4d..5efa9be49e09 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
@@ -658,6 +658,36 @@ public class PersistentTopicsTest extends MockedPulsarServiceBaseTest {
Response.Status.PRECONDITION_FAILED.getStatusCode());
}
+ @Test
+ public void testCreateTopicWithSurroundingWhitespaceIsRejected() {
+ // Clients trim topic names, so a topic created with surrounding whitespace could never be reached.
+ final String topicName = "topic-with-trailing-whitespace ";
+
+ AsyncResponse response = mock(AsyncResponse.class);
+ ArgumentCaptor<RestException> errCaptor = ArgumentCaptor.forClass(RestException.class);
+ persistentTopics.createPartitionedTopic(response, testTenant, testNamespace, topicName, 2, true);
+ verify(response, timeout(5000).times(1)).resume(errCaptor.capture());
+ Assert.assertEquals(errCaptor.getValue().getResponse().getStatus(),
+ Response.Status.PRECONDITION_FAILED.getStatusCode());
+ Assert.assertTrue(errCaptor.getValue().getMessage().contains("must not have leading or trailing whitespace"));
+
+ response = mock(AsyncResponse.class);
+ errCaptor = ArgumentCaptor.forClass(RestException.class);
+ nonPersistentTopic.createPartitionedTopic(response, testTenant, testNamespace, topicName, 2, true);
+ verify(response, timeout(5000).times(1)).resume(errCaptor.capture());
+ Assert.assertEquals(errCaptor.getValue().getResponse().getStatus(),
+ Response.Status.PRECONDITION_FAILED.getStatusCode());
+
+ response = mock(AsyncResponse.class);
+ errCaptor = ArgumentCaptor.forClass(RestException.class);
+ try {
+ persistentTopics.createNonPartitionedTopic(response, testTenant, testNamespace, topicName, true, null);
+ Assert.fail("Should have thrown a RestException");
+ } catch (RestException e) {
+ Assert.assertEquals(e.getResponse().getStatus(), Response.Status.PRECONDITION_FAILED.getStatusCode());
+ }
+ }
+
@SuppressWarnings("deprecation")
@Test
public void testCreatePartitionedTopicHavingNonPartitionTopicWithPartitionSuffix()
@@ -1992,6 +2022,20 @@ public class PersistentTopicsTest extends MockedPulsarServiceBaseTest {
assertEquals(cursor.getCursorProperties().get("foo"), "bar");
}
+ @Test
+ public void testNonPersistentNonPartitionedTopicWithTransactionSuffixIsNotRejected() {
+ // NonPersistentTopics inherits createNonPartitionedTopic from PersistentTopics, so this guards the
+ // non-partitioned non-persistent path against the transaction-internal-name check. If validateCreateTopic
+ // wrongly rejected the transaction-internal suffix here, the async response would resume with an error
+ // status rather than NO_CONTENT.
+ AsyncResponse response = mock(AsyncResponse.class);
+ nonPersistentTopic.createNonPartitionedTopic(response, testTenant, testNamespace,
+ "foo__transaction_pending_ack", true, null);
+ ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
+ verify(response, timeout(5000).times(1)).resume(responseCaptor.capture());
+ Assert.assertEquals(responseCaptor.getValue().getStatus(), Response.Status.NO_CONTENT.getStatusCode());
+ }
+
@Test
public void testPeekMessageWithProperties() throws Exception {
String topicName = "persistent://" + testTenant + "/" + testNamespaceLocal + "/testPeekMessageWithProperties";
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/ScalableTopicsListByPropertyTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/ScalableTopicsListByPropertyTest.java
index 515c1992c51b..fd8253928d6a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/ScalableTopicsListByPropertyTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/ScalableTopicsListByPropertyTest.java
@@ -20,12 +20,21 @@ package org.apache.pulsar.broker.admin;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+import jakarta.ws.rs.client.Client;
+import jakarta.ws.rs.client.ClientBuilder;
+import jakarta.ws.rs.client.Entity;
+import jakarta.ws.rs.core.MediaType;
+import jakarta.ws.rs.core.Response;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import lombok.Cleanup;
import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.common.naming.NamespaceName;
import org.testng.annotations.Test;
/**
@@ -89,4 +98,54 @@ public class ScalableTopicsListByPropertyTest extends SharedPulsarBaseTest {
assertTrue(all.containsAll(Set.of(aliceTopic, bobTopic, carolTopic)),
"expected all three created topics to appear in the unfiltered list, got " + all);
}
+
+ /**
+ * A scalable topic created with trailing whitespace could never be reached: clients trim topic names, so
+ * they would look up the trimmed name instead. The admin client must reject it with a 412
+ * (PreconditionFailedException) on its fast-fail path.
+ */
+ @Test
+ public void testCreateScalableTopicWithSurroundingWhitespaceIsRejected() throws Exception {
+ String topicWithWhitespace = "topic://" + namespace() + "/scalable-with-whitespace-"
+ + UUID.randomUUID().toString().substring(0, 8) + " ";
+ PulsarAdminException.PreconditionFailedException e = expectThrows(
+ PulsarAdminException.PreconditionFailedException.class,
+ () -> admin.scalableTopics().createScalableTopic(topicWithWhitespace, 1,
+ Map.of("owner", "test")));
+ assertTrue(e.getMessage().contains("whitespace"), "expected the surrounding-whitespace rejection, got: "
+ + e.getMessage());
+ }
+
+ /**
+ * The admin client rejects trailing whitespace client-side, so a raw HTTP request is needed to exercise the
+ * server. A percent-encoded trailing space must be decoded by the server and rejected with a 412.
+ */
+ @Test
+ public void testCreateScalableTopicServerSideRejectsEncodedWhitespace() {
+ NamespaceName ns = NamespaceName.get(namespace());
+ String encodedTopic = "scalable-encoded-whitespace-" + UUID.randomUUID().toString().substring(0, 8) + "%20";
+ String url = getWebServiceUrl() + "/admin/v2/scalable/" + ns.getTenant() + "/" + ns.getLocalName()
+ + "/" + encodedTopic;
+ @Cleanup
+ Client client = ClientBuilder.newClient();
+ try (Response response = client.target(url)
+ .request(MediaType.APPLICATION_JSON)
+ .put(Entity.json(Map.of("owner", "test")))) {
+ assertEquals(response.getStatus(), Response.Status.PRECONDITION_FAILED.getStatusCode());
+ assertTrue(response.readEntity(String.class).contains("whitespace"));
+ }
+ }
+
+ /**
+ * The transaction-internal-name rule is specific to persistent topics. The shared whitespace validation
+ * must not extend it to non-persistent or scalable topics.
+ */
+ @Test
+ public void testTransactionInternalSuffixIsNotRejectedForNonPersistentOrScalableTopics() throws Exception {
+ String ns = namespace();
+ admin.nonPersistentTopics().createPartitionedTopic(
+ "non-persistent://" + ns + "/foo__transaction_pending_ack", 1);
+ admin.scalableTopics().createScalableTopic(
+ "topic://" + ns + "/foo__transaction_pending_ack", 1, Map.of("owner", "test"));
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java
index f8b672489fd5..65274f1735f9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java
@@ -25,8 +25,10 @@ import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import java.util.List;
import java.util.Map;
+import java.util.NoSuchElementException;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import lombok.Cleanup;
@@ -47,6 +49,7 @@ import org.apache.pulsar.common.policies.data.InactiveTopicDeleteMode;
import org.apache.pulsar.common.policies.data.InactiveTopicPolicies;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.policies.data.TopicType;
+import org.apache.pulsar.common.util.FutureUtil;
import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
@@ -87,6 +90,28 @@ public class BrokerServiceAutoTopicCreationTest extends BrokerTestBase{
assertFalse(admin.topics().getPartitionedTopicList("prop/ns-abc").contains(topicString));
}
+ @Test
+ public void testAutoTopicCreationRejectsSurroundingWhitespace() throws Exception {
+ pulsar.getConfiguration().setAllowAutoTopicCreation(true);
+ pulsar.getConfiguration().setAllowAutoTopicCreationType(TopicType.NON_PARTITIONED);
+
+ // Go straight to the broker service: the Java client trims topic names, so it could never send this name.
+ final String topicString = "persistent://prop/ns-abc/auto-created-with-whitespace ";
+ assertFalse(pulsar.getBrokerService().isAllowAutoTopicCreationAsync(topicString).get());
+
+ try {
+ pulsar.getBrokerService().getOrCreateTopic(topicString).get();
+ fail("Should not have auto-created a topic whose name has surrounding whitespace");
+ } catch (ExecutionException e) {
+ // Auto-creation is denied, so getTopic returns Optional.empty() and getOrCreateTopic's
+ // thenApply(Optional::get) throws NoSuchElementException for that empty optional.
+ Throwable root = FutureUtil.unwrapCompletionException(e);
+ assertTrue(root instanceof NoSuchElementException,
+ "expected NoSuchElementException because the topic was not auto-created, got: " + root);
+ }
+ assertFalse(admin.namespaces().getTopics("prop/ns-abc").contains(topicString));
+ }
+
@Test
public void testAutoNonPartitionedTopicCreationOnProduce() throws Exception{
pulsar.getConfiguration().setAllowAutoTopicCreation(true);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerCreationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerCreationTest.java
index b4199c9ef08c..8730a37f3e7a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerCreationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerCreationTest.java
@@ -19,7 +19,11 @@
package org.apache.pulsar.client.api;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
import lombok.Cleanup;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.PulsarClientException.NotAllowedException;
@@ -54,6 +58,38 @@ public class ConsumerCreationTest extends ProducerConsumerBase {
};
}
+ /**
+ * A topic name with trailing whitespace used to be accepted on creation, but clients trim topic names, so the
+ * consumer looked up the trimmed name and failed with a confusing TopicDoesNotExistException. Creation must be
+ * rejected instead, and the trimmed name must keep working.
+ */
+ @Test
+ public void testCreatePartitionedTopicWithTrailingWhitespaceIsRejected() throws Exception {
+ conf.setAllowAutoTopicCreation(false);
+
+ String trimmedTopic = "persistent://public/default/testCreatePartitionedTopicWithTrailingWhitespace";
+ String topicWithWhitespace = trimmedTopic + " ";
+
+ PulsarAdminException e1 = expectThrows(PulsarAdminException.class,
+ () -> admin.topics().createPartitionedTopic(topicWithWhitespace, 2));
+ assertTrue(e1.getMessage().contains("whitespace"),
+ "expected the surrounding-whitespace rejection, got: " + e1.getMessage());
+ PulsarAdminException e2 = expectThrows(PulsarAdminException.class,
+ () -> admin.topics().createNonPartitionedTopic(topicWithWhitespace));
+ assertTrue(e2.getMessage().contains("whitespace"),
+ "expected the surrounding-whitespace rejection, got: " + e2.getMessage());
+
+ // No partial metadata is left behind by the rejected creation.
+ assertFalse(admin.topics().getPartitionedTopicList("public/default").contains(topicWithWhitespace));
+
+ // The trimmed name is what the client resolves to, so creating it makes the original call site work.
+ admin.topics().createPartitionedTopic(trimmedTopic, 2);
+ @Cleanup
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicWithWhitespace)
+ .subscriptionName("my-sub").subscribe();
+ assertEquals(consumer.getTopic(), trimmedTopic);
+ }
+
@Test(dataProvider = "topicDomainProvider")
public void testCreateConsumerWhenTopicTypeMismatch(TopicDomain domain)
throws PulsarAdminException, PulsarClientException {
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NonPersistentTopicsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NonPersistentTopicsImpl.java
index ebd332ab546b..042d98b0416b 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NonPersistentTopicsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NonPersistentTopicsImpl.java
@@ -22,16 +22,19 @@ import static com.google.common.base.Preconditions.checkArgument;
import jakarta.ws.rs.client.Entity;
import jakarta.ws.rs.client.WebTarget;
import jakarta.ws.rs.core.MediaType;
+import jakarta.ws.rs.core.Response.Status;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import org.apache.pulsar.client.admin.NonPersistentTopics;
import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.admin.PulsarAdminException.PreconditionFailedException;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.policies.data.NonPersistentTopicStats;
import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
+import org.apache.pulsar.common.util.FutureUtil;
@SuppressWarnings("deprecation")
public class NonPersistentTopicsImpl extends BaseResource implements NonPersistentTopics {
@@ -52,6 +55,12 @@ public class NonPersistentTopicsImpl extends BaseResource implements NonPersiste
public CompletableFuture<Void> createPartitionedTopicAsync(String topic, int numPartitions) {
checkArgument(numPartitions > 0, "Number of partitions should be more than 0");
TopicName topicName = validateTopic(topic);
+ try {
+ TopicName.validateTopicNameForCreation(topicName);
+ } catch (IllegalArgumentException e) {
+ return FutureUtil.failedFuture(
+ new PreconditionFailedException(e, e.getMessage(), Status.PRECONDITION_FAILED.getStatusCode()));
+ }
WebTarget path = topicPath(topicName, "partitions");
return asyncPutRequest(path, Entity.entity(numPartitions, MediaType.APPLICATION_JSON));
}
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ScalableTopicsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ScalableTopicsImpl.java
index e4905aa63c82..068101ad6e9d 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ScalableTopicsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ScalableTopicsImpl.java
@@ -22,10 +22,12 @@ import jakarta.ws.rs.client.Entity;
import jakarta.ws.rs.client.WebTarget;
import jakarta.ws.rs.core.GenericType;
import jakarta.ws.rs.core.MediaType;
+import jakarta.ws.rs.core.Response.Status;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.admin.PulsarAdminException.PreconditionFailedException;
import org.apache.pulsar.client.admin.ScalableTopics;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.common.naming.NamespaceName;
@@ -33,6 +35,7 @@ import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.AutoScalePolicyOverride;
import org.apache.pulsar.common.policies.data.ScalableTopicMetadata;
import org.apache.pulsar.common.policies.data.ScalableTopicStats;
+import org.apache.pulsar.common.util.FutureUtil;
public class ScalableTopicsImpl extends BaseResource implements ScalableTopics {
private final WebTarget adminScalable;
@@ -108,6 +111,12 @@ public class ScalableTopicsImpl extends BaseResource implements ScalableTopics {
public CompletableFuture<Void> createScalableTopicAsync(String topic, int numInitialSegments,
Map<String, String> properties) {
TopicName tn = validateTopic(topic);
+ try {
+ TopicName.validateTopicNameForCreation(tn);
+ } catch (IllegalArgumentException e) {
+ return FutureUtil.failedFuture(
+ new PreconditionFailedException(e, e.getMessage(), Status.PRECONDITION_FAILED.getStatusCode()));
+ }
WebTarget path = topicPath(tn).queryParam("numInitialSegments", numInitialSegments);
Entity<?> entity = (properties != null && !properties.isEmpty())
? Entity.entity(properties, MediaType.APPLICATION_JSON)
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
index 03666134b5e5..c4ebb5bbcd94 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
@@ -56,6 +56,7 @@ import org.apache.pulsar.client.admin.LongRunningProcessStatus;
import org.apache.pulsar.client.admin.OffloadProcessStatus;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.admin.PulsarAdminException.NotFoundException;
+import org.apache.pulsar.client.admin.PulsarAdminException.PreconditionFailedException;
import org.apache.pulsar.client.admin.Topics;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.Message;
@@ -100,6 +101,7 @@ import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.stats.AnalyzeSubscriptionBacklogResult;
import org.apache.pulsar.common.util.Codec;
import org.apache.pulsar.common.util.DateFormatter;
+import org.apache.pulsar.common.util.FutureUtil;
@SuppressWarnings("deprecation")
@CustomLog
@@ -330,6 +332,12 @@ public class TopicsImpl extends BaseResource implements Topics {
@Override
public CompletableFuture<Void> createNonPartitionedTopicAsync(String topic, Map<String, String> properties){
TopicName tn = validateTopic(topic);
+ try {
+ TopicName.validateTopicNameForCreation(tn);
+ } catch (IllegalArgumentException e) {
+ return FutureUtil.failedFuture(
+ new PreconditionFailedException(e, e.getMessage(), Status.PRECONDITION_FAILED.getStatusCode()));
+ }
WebTarget path = topicPath(tn);
properties = properties == null ? new HashMap<>() : properties;
return asyncPutRequest(path, Entity.entity(properties, MediaType.APPLICATION_JSON));
@@ -346,6 +354,12 @@ public class TopicsImpl extends BaseResource implements Topics {
String topic, int numPartitions, boolean createLocalTopicOnly, Map<String, String> properties) {
checkArgument(numPartitions > 0, "Number of partitions should be more than 0");
TopicName tn = validateTopic(topic);
+ try {
+ TopicName.validateTopicNameForCreation(tn);
+ } catch (IllegalArgumentException e) {
+ return FutureUtil.failedFuture(
+ new PreconditionFailedException(e, e.getMessage(), Status.PRECONDITION_FAILED.getStatusCode()));
+ }
WebTarget path = topicPath(tn, "partitions")
.queryParam("createLocalTopicOnly", Boolean.toString(createLocalTopicOnly));
Entity entity;
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
index ea70d11a0ac7..7ae2cbe3a517 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
@@ -107,6 +107,37 @@ public class TopicName implements ServiceUnitId {
}
}
+ /**
+ * Checks whether a topic name can be used to create a new topic.
+ *
+ * <p>Pulsar clients trim the <em>entire</em> topic string before resolving it (see
+ * {@link org.apache.pulsar.client.api.ConsumerBuilder#topic(String...)}), so a topic whose full name has
+ * surrounding whitespace would be created under one name but addressed under the trimmed name. Checking the
+ * full name (rather than only the local name) matches that client behavior: a leading space in the local name
+ * of a fully-qualified name sits in the middle of the string and is not trimmed by clients, so it stays
+ * allowed.
+ *
+ * <p>This is only meant to be checked on topic creation paths, so topics which already have such a name
+ * remain listable, readable and deletable.
+ */
+ public static boolean isValidForCreation(TopicName topicName) {
+ String topic = topicName.toString();
+ return topic.equals(StringUtils.trim(topic));
+ }
+
+ /**
+ * Validates that a topic name can be used to create a new topic.
+ *
+ * @throws IllegalArgumentException if the topic name cannot be used to create a topic
+ * @see #isValidForCreation(TopicName)
+ */
+ public static void validateTopicNameForCreation(TopicName topicName) {
+ if (!isValidForCreation(topicName)) {
+ throw new IllegalArgumentException("Invalid topic name: '" + topicName
+ + "'. Topic name must not have leading or trailing whitespace.");
+ }
+ }
+
public static String getPartitionPattern(String topic) {
return "^" + Pattern.quote(get(topic).getPartitionedTopicName().toString()) + "-partition-[0-9]+$";
}
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
index eb9254ecf4a1..71b2a2839d55 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
@@ -194,6 +194,41 @@ public class TopicNameTest {
assertTrue(TopicName.get("mytopic-partition-12").isPartitioned());
}
+ @Test
+ public void testValidateTopicNameForCreation() {
+ // A trailing space in the full name is trimmed by clients, so the topic would be created under one
+ // name and addressed under the trimmed one. Only trailing whitespace is unreachable this way: a
+ // leading space in the local name sits mid-string in a fully-qualified name and is not trimmed.
+ String[] invalidNames = {
+ "persistent://myprop/myns/mytopic ",
+ "persistent://myprop/myns/mytopic\t",
+ "persistent://myprop/myns/mytopic\n",
+ "non-persistent://myprop/myns/mytopic ",
+ // leading space is allowed, but a trailing space is still trimmed away
+ "persistent://myprop/myns/ mytopic ",
+ };
+ for (String invalidName : invalidNames) {
+ TopicName topicName = TopicName.get(invalidName);
+ assertFalse(TopicName.isValidForCreation(topicName), invalidName);
+ assertThrows(IllegalArgumentException.class,
+ () -> TopicName.validateTopicNameForCreation(topicName));
+ }
+
+ // Whitespace inside the local name (including a leading space) round-trips correctly and stays allowed.
+ String[] validNames = {
+ "persistent://myprop/myns/my topic",
+ "persistent://myprop/myns/mytopic",
+ "persistent://myprop/myns/mytopic-partition-0",
+ "non-persistent://myprop/myns/my topic",
+ "persistent://myprop/myns/ mytopic",
+ };
+ for (String validName : validNames) {
+ TopicName topicName = TopicName.get(validName);
+ assertTrue(TopicName.isValidForCreation(topicName), validName);
+ TopicName.validateTopicNameForCreation(topicName);
+ }
+ }
+
@Test
public void testDecodeEncode() throws Exception {
String encodedName = "a%3Aen-in_in_business_content_item_20150312173022_https%5C%3A";