[GitHub] [pulsar] sijie commented on a change in pull request #6478: Fix create partitioned topic with a substring of an existing topic name.

[GitBox <gi...@apache.org>]

sijie commented on a change in pull request #6478: Fix create partitioned topic with a substring of an existing topic name.
URL: https://github.com/apache/pulsar/pull/6478#discussion_r388655914
 
 

 ##########
 File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
 ##########
 @@ -707,4 +709,93 @@ protected boolean isNamespaceReplicated(NamespaceName namespaceName) {
         partitionedTopics.sort(null);
         return partitionedTopics;
     }
+
+    protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int numPartitions) {
+        try {
+            validateAdminAccessForTenant(topicName.getTenant());
+        } catch (Exception e) {
+            log.error("[{}] Failed to create partitioned topic {}", clientAppId(), topicName, e);
+            resumeAsyncResponseExceptionally(asyncResponse, e);
+            return;
+        }
+        if (numPartitions <= 0) {
+            asyncResponse.resume(new RestException(Status.NOT_ACCEPTABLE, "Number of partitions should be more than 0"));
+            return;
+        }
+        checkTopicExistsAsync(topicName).thenAccept(exists -> {
+            if (exists) {
+                log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
+                asyncResponse.resume(new RestException(Status.CONFLICT, "This topic already exists"));
+            } else {
+                try {
+                    String path = ZkAdminPaths.partitionedTopicPath(topicName);
+                    byte[] data = jsonMapper().writeValueAsBytes(new PartitionedTopicMetadata(numPartitions));
+                    zkCreateOptimisticAsync(globalZk(), path, data, (rc, s, o, s1) -> {
+                        if (KeeperException.Code.OK.intValue() == rc) {
+                            if (topicName.isPersistent()) {
+                                tryCreatePartitionsAsync(numPartitions);
+                            }
+                            globalZk().sync(path, (rc2, s2, ctx) -> {
 
 Review comment:
   This should happen in the callback of `tryCreatePartitionAsync`.

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services