新建订阅

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

@@ -0,0 +1,17 @@
+package com.xuxd.kafka.console.beans.dto;
+
+import lombok.Data;
+
+/**
+ * kafka-console-ui.
+ *
+ * @author xuxd
+ * @date 2021-10-21 15:43:56
+ **/
+@Data
+public class AddSubscriptionDTO {
+
+    private String groupId;
+
+    private String topic;
+}

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

@@ -1,5 +1,6 @@
 package com.xuxd.kafka.console.controller;
 
+import com.xuxd.kafka.console.beans.dto.AddSubscriptionDTO;
 import com.xuxd.kafka.console.beans.dto.QueryConsumerGroupDTO;
 import com.xuxd.kafka.console.service.ConsumerService;
 import java.util.Collections;
@@ -60,4 +61,9 @@ public class ConsumerController {
     public Object getConsumerDetail(@RequestParam String groupId) {
         return consumerService.getConsumerDetail(groupId);
     }
+
+    @PostMapping("/subscription")
+    public Object addSubscription(@RequestBody AddSubscriptionDTO subscriptionDTO) {
+        return consumerService.addSubscription(subscriptionDTO.getGroupId(), subscriptionDTO.getTopic());
+    }
 }

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

@@ -30,6 +30,11 @@ public class TopicController {
     @Autowired
     private TopicService topicService;
 
+    @GetMapping
+    public Object getTopicNameList() {
+        return topicService.getTopicNameList(false);
+    }
+
     @GetMapping("/list")
     public Object getTopicList(@RequestParam(required = false) String topic, @RequestParam String type) {
         return topicService.getTopicList(topic, TopicType.valueOf(type.toUpperCase()));

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

@@ -20,4 +20,6 @@ public interface ConsumerService {
     ResponseData getConsumerMembers(String groupId);
 
     ResponseData getConsumerDetail(String groupId);
+
+    ResponseData addSubscription(String groupId, String topic);
 }

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

@@ -15,7 +15,7 @@ import org.apache.kafka.clients.admin.NewTopic;
  **/
 public interface TopicService {
 
-    ResponseData getTopicNameList();
+    ResponseData getTopicNameList(boolean internal);
 
     ResponseData getTopicList(String topic, TopicType type);
 

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

@@ -98,4 +98,22 @@ public class ConsumerServiceImpl implements ConsumerService {
         });
         return ResponseData.create().data(res).success();
     }
+
+    @Override public ResponseData addSubscription(String groupId, String topic) {
+        // check whether exist subscription relationship.
+        Collection<ConsumerConsole.TopicPartitionConsumeInfo> consumerDetail = consumerConsole.getConsumerDetail(Collections.singleton(groupId));
+        if (CollectionUtils.isNotEmpty(consumerDetail)) {
+            List<ConsumerConsole.TopicPartitionConsumeInfo> collect = consumerDetail.stream()
+                .filter(t -> t.getTopicPartition().topic().equals(topic)).collect(Collectors.toList());
+            if (CollectionUtils.isNotEmpty(collect)) {
+                return ResponseData.create().failed("The subscription exist.");
+            }
+        }
+
+        // consumer message and commit offset.
+
+
+        // reset consume offset to 0.
+        return ResponseData.create().success();
+    }
 }

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

@@ -38,8 +38,8 @@ public class TopicServiceImpl implements TopicService {
     @Autowired
     private TopicConsole topicConsole;
 
-    @Override public ResponseData getTopicNameList() {
-        return ResponseData.create().data(topicConsole.getTopicNameList(true)).success();
+    @Override public ResponseData getTopicNameList(boolean internal) {
+        return ResponseData.create().data(topicConsole.getTopicNameList(internal)).success();
     }
 
     @Override public ResponseData getTopicList(String topic, TopicType type) {