[fix][broker]V1 topic creation should fail if namespace does not exist

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

@@ -323,7 +323,12 @@ protected void internalRevokePermissionsOnTopic(AsyncResponse asyncResponse, Str
 
     protected CompletableFuture<Void> internalCreateNonPartitionedTopicAsync(boolean authoritative,
                                                      Map<String, String> properties) {
-        CompletableFuture<Void> ret = validateNonPartitionTopicNameAsync(topicName.getLocalName());
+        CompletableFuture<Void> ret = namespaceResources().namespaceExistsAsync(namespaceName)
+            .thenAccept(exists -> {
+                if (!exists) {
+                    throw new RestException(Status.NOT_FOUND, "V1 namespace [" + namespaceName + "] does not exist");
+                }
+            }). thenCompose(__ -> validateNonPartitionTopicNameAsync(topicName.getLocalName()));
         if (topicName.isGlobal()) {
             ret = ret.thenCompose(__ -> validateGlobalNamespaceOwnershipAsync(namespaceName));
         }

pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java

@@ -1074,7 +1074,12 @@ && pulsar().getBrokerService().isAuthorizationEnabled()) {
                         }
                     });
         }
-        return CompletableFuture.completedFuture(null);
+        return namespaceResources().namespaceExistsAsync(namespaceName)
+            .thenAccept(exists -> {
+                if (!exists) {
+                    throw new RestException(Status.NOT_FOUND, "V1 namespace [" + namespaceName + "] does not exist");
+                }
+            });
     }
 
     public void validateNamespacePolicyOperation(NamespaceName namespaceName, PolicyName policy,

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

@@ -34,6 +34,7 @@
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.impl.ClientCnx;
@@ -635,4 +636,37 @@ public void testTenantNotExist(TopicDomain topicDomain) throws Exception {
             assertEquals(getLookupRequestPermits(), lookupPermitsBefore);
         });
     }
+
+    @Test
+    public void testTopicNameContainsSpecialCharacters() throws Exception {
+        // V1 topic create failed because v1 namespace does not exist.
+        String topic = "persistent://public/default/tp/h";
+        try {
+            admin1.topics().createNonPartitionedTopic(topic);
+            fail("Expected a namespace not found ex, since the v1 namespace does not exist");
+        } catch (PulsarAdminException.NotFoundException ex) {
+            assertTrue(ex.getMessage().contains("does not exist"));
+        }
+        try {
+            admin1.topics().createPartitionedTopic(topic, 1);
+            fail("Expected a namespace not found ex, since the v1 namespace does not exist");
+        } catch (PulsarAdminException.NotFoundException ex) {
+            assertTrue(ex.getMessage().contains("does not exist"));
+        }
+
+        // The topic can be created after v1 namespace was created
+        String v1Ns = "public/" + pulsar1.getConfig().getClusterName() + "/default";
+        admin1.namespaces().createNamespace(v1Ns);
+        String topic2 = "persistent://" + v1Ns + "/h";
+        String topic3 = "persistent://" + v1Ns + "/h2";
+        admin1.topics().createNonPartitionedTopic(topic2);
+        admin1.topics().createPartitionedTopic(topic3, 1);
+        List<String> v1Topics = admin1.topics().getList(v1Ns);
+        assertTrue(v1Topics.contains(topic2));
+        assertTrue(v1Topics.contains(topic3 + "-partition-0"));
+
+        // cleanup.
+        admin1.topics().delete(topic2, false);
+        admin1.topics().deletePartitionedTopic(topic3, false);
+    }
 }