新增topic

src/main/java/com/xuxd/kafka/console/beans/dto/NewTopicDTO.java

@@ -0,0 +1,30 @@
+package com.xuxd.kafka.console.beans.dto;
+
+import java.util.HashMap;
+import java.util.Map;
+import lombok.Data;
+import org.apache.commons.collections.MapUtils;
+import org.apache.kafka.clients.admin.NewTopic;
+
+/**
+ * kafka-console-ui.
+ *
+ * @author xuxd
+ * @date 2021-10-13 14:58:11
+ **/
+@Data
+public class NewTopicDTO {
+
+    private String name;
+    private Integer numPartitions;
+    private Short replicationFactor;
+    private Map<String, String> configs = new HashMap<>();
+
+    public NewTopic toNewTopic() {
+        NewTopic topic = new NewTopic(name, numPartitions, replicationFactor);
+        if (MapUtils.isNotEmpty(configs)) {
+            topic.configs(configs);
+        }
+        return topic;
+    }
+}

src/main/java/com/xuxd/kafka/console/controller/TopicController.java

@@ -1,10 +1,13 @@
 package com.xuxd.kafka.console.controller;
 
+import com.xuxd.kafka.console.beans.dto.NewTopicDTO;
 import com.xuxd.kafka.console.beans.enums.TopicType;
 import com.xuxd.kafka.console.service.TopicService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.DeleteMapping;
 import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
@@ -36,4 +39,9 @@ public class TopicController {
     public Object getTopicPartitionInfo(@RequestParam String topic) {
         return topicService.getTopicPartitionInfo(topic);
     }
+
+    @PostMapping("/new")
+    public Object createNewTopic(@RequestBody NewTopicDTO topicDTO) {
+        return topicService.createTopic(topicDTO.toNewTopic());
+    }
 }

src/main/java/com/xuxd/kafka/console/service/TopicService.java

@@ -2,6 +2,7 @@ package com.xuxd.kafka.console.service;
 
 import com.xuxd.kafka.console.beans.ResponseData;
 import com.xuxd.kafka.console.beans.enums.TopicType;
+import org.apache.kafka.clients.admin.NewTopic;
 
 /**
  * kafka-console-ui.
@@ -18,4 +19,6 @@ public interface TopicService {
     ResponseData deleteTopic(String topic);
 
     ResponseData getTopicPartitionInfo(String topic);
+
+    ResponseData createTopic(NewTopic topic);
 }

src/main/java/com/xuxd/kafka/console/service/impl/TopicServiceImpl.java

@@ -16,6 +16,7 @@ import java.util.stream.Collectors;
 import kafka.console.TopicConsole;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.TopicPartitionInfo;
@@ -106,4 +107,9 @@ public class TopicServiceImpl implements TopicService {
         }
         return ResponseData.create().data(voList).success();
     }
+
+    @Override public ResponseData createTopic(NewTopic topic) {
+        Tuple2<Object, String> createResult = topicConsole.createTopic(topic);
+        return (boolean) createResult._1 ? ResponseData.create().success() : ResponseData.create().failed(String.valueOf(createResult._2));
+    }
 }