feat: expose create topic API to ProxyMetadataService (#503)

Signed-off-by: Li Zhanhui <[email protected]>
AutoMQ · Oct 30, 2023 · 8ebaf69 · 8ebaf69
1 parent 1f7e671
commit 8ebaf69
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 0 deletions.
diff --git a/metadata/src/main/java/com/automq/rocketmq/metadata/DefaultProxyMetadataService.java b/metadata/src/main/java/com/automq/rocketmq/metadata/DefaultProxyMetadataService.java
@@ -18,6 +18,7 @@
 package com.automq.rocketmq.metadata;
 
 import apache.rocketmq.controller.v1.ConsumerGroup;
+import apache.rocketmq.controller.v1.CreateTopicRequest;
 import apache.rocketmq.controller.v1.MessageQueueAssignment;
 import apache.rocketmq.controller.v1.Topic;
 import com.automq.rocketmq.controller.metadata.MetadataStore;
@@ -38,6 +39,12 @@ public DefaultProxyMetadataService(MetadataStore metadataStore) {
         this.metadataStore = metadataStore;
     }
 
+    @Override
+    public CompletableFuture<Topic> createTopic(CreateTopicRequest request) {
+        return metadataStore.createTopic(request)
+            .thenComposeAsync(this::topicOf);
+    }
+
     @Override
     public CompletableFuture<Topic> topicOf(String topicName) {
         return topicOf(null, topicName);

diff --git a/metadata/src/main/java/com/automq/rocketmq/metadata/api/ProxyMetadataService.java b/metadata/src/main/java/com/automq/rocketmq/metadata/api/ProxyMetadataService.java
@@ -18,12 +18,16 @@
 package com.automq.rocketmq.metadata.api;
 
 import apache.rocketmq.controller.v1.ConsumerGroup;
+import apache.rocketmq.controller.v1.CreateTopicRequest;
 import apache.rocketmq.controller.v1.MessageQueueAssignment;
 import apache.rocketmq.controller.v1.Topic;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 
 public interface ProxyMetadataService {
+
+    CompletableFuture<Topic> createTopic(CreateTopicRequest request);
+
     /**
      * Query the topic metadata of a given topic name.
      *

diff --git a/proxy/src/test/java/com/automq/rocketmq/proxy/mock/MockProxyMetadataService.java b/proxy/src/test/java/com/automq/rocketmq/proxy/mock/MockProxyMetadataService.java
@@ -19,6 +19,7 @@
 
 import apache.rocketmq.controller.v1.AcceptTypes;
 import apache.rocketmq.controller.v1.ConsumerGroup;
+import apache.rocketmq.controller.v1.CreateTopicRequest;
 import apache.rocketmq.controller.v1.MessageQueue;
 import apache.rocketmq.controller.v1.MessageQueueAssignment;
 import apache.rocketmq.controller.v1.MessageType;
@@ -32,6 +33,11 @@
 public class MockProxyMetadataService implements ProxyMetadataService {
     Map<Long, Long> offsetMap = new HashMap<>();
 
+    @Override
+    public CompletableFuture<Topic> createTopic(CreateTopicRequest request) {
+        return topicOf(request.getTopic());
+    }
+
     @Override
     public CompletableFuture<Topic> topicOf(String topicName) {
         // Return a dummy topic