[fix][admin] Listen partitioned topic creation event (apache#23680)

Signed-off-by: Zixuan Liu <[email protected]> (cherry picked from commit 0a2ffe4) (cherry picked from commit 7dcc255)
datastax · Dec 12, 2024 · d8bc659 · d8bc659
1 parent 74fefda
commit d8bc659
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 6 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
@@ -45,6 +45,8 @@
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.authorization.AuthorizationService;
 import org.apache.pulsar.broker.resources.ClusterResources;
+import org.apache.pulsar.broker.service.TopicEventsListener.EventStage;
+import org.apache.pulsar.broker.service.TopicEventsListener.TopicEvent;
 import org.apache.pulsar.broker.service.plugin.InvalidEntryFilterException;
 import org.apache.pulsar.broker.web.PulsarWebResource;
 import org.apache.pulsar.broker.web.RestException;
@@ -163,6 +165,10 @@ public CompletableFuture<Void> validatePoliciesReadOnlyAccessAsync() {
 
     protected CompletableFuture<Void> tryCreatePartitionsAsync(int numPartitions) {
         if (!topicName.isPersistent()) {
+            for (int i = 0; i < numPartitions; i++) {
+                pulsar().getBrokerService().getTopicEventsDispatcher()
+                        .notify(topicName.getPartition(i).toString(), TopicEvent.CREATE, EventStage.SUCCESS);
+            }
             return CompletableFuture.completedFuture(null);
         }
         List<CompletableFuture<Void>> futures = new ArrayList<>(numPartitions);
@@ -198,6 +204,8 @@ private CompletableFuture<Void> tryCreatePartitionAsync(final int partition) {
                     }
                     return null;
                 });
+        pulsar().getBrokerService().getTopicEventsDispatcher()
+                .notifyOnCompletion(result, topicName.getPartition(partition).toString(), TopicEvent.CREATE);
         return result;
     }
 
@@ -599,6 +607,13 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
                         throw new RestException(Status.CONFLICT, "This topic already exists");
                     }
                 })
+                .thenRun(() -> {
+                    for (int i = 0; i < numPartitions; i++) {
+                        pulsar().getBrokerService().getTopicEventsDispatcher()
+                                .notify(topicName.getPartition(i).toString(), TopicEvent.CREATE,
+                                        EventStage.BEFORE);
+                    }
+                })
                 .thenCompose(__ -> provisionPartitionedTopicPath(numPartitions, createLocalTopicOnly, properties))
                 .thenCompose(__ -> tryCreatePartitionsAsync(numPartitions))
                 .thenRun(() -> {

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/TopicEventsListenerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/TopicEventsListenerTest.java
@@ -265,12 +265,23 @@ public void testTopicAutoGC(String topicTypePersistence, String topicTypePartiti
     private void createTopicAndVerifyEvents(String topicDomain, String topicTypePartitioned, String topicName) throws Exception {
         final String[] expectedEvents;
         if (topicDomain.equalsIgnoreCase("persistent") || topicTypePartitioned.equals("partitioned")) {
-            expectedEvents = new String[]{
-                    "LOAD__BEFORE",
-                    "CREATE__BEFORE",
-                    "CREATE__SUCCESS",
-                    "LOAD__SUCCESS"
-            };
+            if (topicTypePartitioned.equals("partitioned")) {
+                expectedEvents = new String[]{
+                        "CREATE__BEFORE",
+                        "CREATE__SUCCESS",
+                        "LOAD__BEFORE",
+                        "CREATE__BEFORE",
+                        "CREATE__SUCCESS",
+                        "LOAD__SUCCESS"
+                };
+            } else {
+                expectedEvents = new String[]{
+                        "LOAD__BEFORE",
+                        "CREATE__BEFORE",
+                        "CREATE__SUCCESS",
+                        "LOAD__SUCCESS"
+                };
+            }
         } else {
             expectedEvents = new String[]{
                     // Before https://github.com/apache/pulsar/pull/21995, Pulsar will skip create topic if the topic