[fix][broker] Reject topic names with leading or trailing whitespace on creation

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java

@@ -56,6 +56,7 @@
 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,27 @@ protected void validateTopicName(String tenant, String namespace, String encoded
         }
     }
 
+    /**
+     * 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, and lets future create-time checks apply
+     * uniformly to all topic types.
+     */
+    protected void validateCreateTopic(TopicName topicName) {
+        if (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) {

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

@@ -224,13 +224,6 @@ protected CompletableFuture<Map<String, Set<AuthAction>>> internalGetPermissions
                 .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);

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java

@@ -220,6 +220,7 @@ protected void validateAndCreatePartitionedTopic(AsyncResponse asyncResponse, St
             validateNamespaceName(tenant, namespace);
             validateGlobalNamespaceOwnership();
             validateTopicName(tenant, namespace, encodedTopic);
+            validateCreateTopic(topicName);
             internalCreatePartitionedTopic(asyncResponse, numPartitions, createLocalTopicOnly, properties);
         } catch (Exception e) {
             log.error()

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/ScalableTopics.java

@@ -179,6 +179,7 @@ public void createScalableTopic(
             Map<String, String> properties) {
         validateNamespaceName(tenant, namespace);
         TopicName tn = TopicName.get(TopicDomain.topic.value(), namespaceName, encodedTopic);
+        validateCreateTopic(tn);
 
         validateNamespaceOperationAsync(namespaceName, NamespaceOperation.CREATE_TOPIC)
                 .thenCompose(__ -> {

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -4068,6 +4068,17 @@ private CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(final TopicName
             return CompletableFuture.completedFuture(false);
         }
 
+        // A topic whose local name has leading or trailing 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 local name has leading or trailing 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.

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java

@@ -658,6 +658,36 @@ public void testCreateNonPartitionedTopicWithInvalidName() {
                 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()

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/ScalableTopicsListByPropertyTest.java

@@ -20,12 +20,14 @@
 
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
 import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
+import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.testng.annotations.Test;
 
 /**
@@ -89,4 +91,21 @@ public void listScalableTopicsFilteredByProperty() throws Exception {
         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 surrounding whitespace could never be reached: clients trim topic names, so
+     * they would look up the trimmed name instead. Creation must be rejected with a 412 (PreconditionFailedException)
+     * on both the client-side fast-fail path and the server-side validation in ScalableTopics.createScalableTopic.
+     */
+    @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());
+    }
 }

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java

@@ -25,8 +25,10 @@
 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.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 void testAutoNonPartitionedTopicCreation() throws Exception{
         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);

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 Object[][] topicDomainProvider() {
         };
     }
 
+    /**
+     * 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 {

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NonPersistentTopicsImpl.java

@@ -22,16 +22,19 @@
 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 void createPartitionedTopic(String topic, int numPartitions) throws Pulsa
     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));
     }

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/ScalableTopicsImpl.java

@@ -22,17 +22,20 @@
 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;
 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 void createScalableTopic(String topic, int numInitialSegments, Map<String
     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)

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java

@@ -56,6 +56,7 @@
 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.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 void createMissedPartitions(String topic) throws PulsarAdminException {
     @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 CompletableFuture<Void> createPartitionedTopicAsync(
             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;