Topic -> 消费详情

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

@@ -116,4 +116,9 @@ public class ConsumerController {
     public Object getSubscribeTopicList(@RequestParam String groupId) {
         return consumerService.getSubscribeTopicList(groupId);
     }
+
+    @GetMapping("/topic/subscribed")
+    public Object getTopicSubscribedByGroups(@RequestParam String topic) {
+        return consumerService.getTopicSubscribedByGroups(topic);
+    }
 }

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

@@ -32,4 +32,6 @@ public interface ConsumerService {
     ResponseData getGroupIdList();
 
     ResponseData getSubscribeTopicList(String groupId);
+
+    ResponseData getTopicSubscribedByGroups(String topic);
 }

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

@@ -38,6 +38,8 @@ public class ConsumerServiceImpl implements ConsumerService {
     @Autowired
     private ConsumerConsole consumerConsole;
 
+    private TopicSubscribedInfo topicSubscribedInfo = new TopicSubscribedInfo();
+
     @Override public ResponseData getConsumerGroupList(List<String> groupIds, Set<ConsumerGroupState> states) {
         String simulateGroup = "inner_xxx_not_exit_group_###" + System.currentTimeMillis();
         Set<String> groupList = new HashSet<>();
@@ -142,4 +144,63 @@ public class ConsumerServiceImpl implements ConsumerService {
     @Override public ResponseData getSubscribeTopicList(String groupId) {
         return ResponseData.create().data(consumerConsole.listSubscribeTopics(groupId).keySet()).success();
     }
+
+    @Override public ResponseData getTopicSubscribedByGroups(String topic) {
+        if (topicSubscribedInfo.isNeedRefresh(topic)) {
+            Set<String> groupIdList = consumerConsole.getConsumerGroupIdList(Collections.emptySet());
+            Map<String, Set<String>> cache = new HashMap<>();
+            Map<String, List<TopicPartition>> subscribeTopics = consumerConsole.listSubscribeTopics(groupIdList);
+
+            subscribeTopics.forEach((groupId, tl) -> {
+                tl.forEach(topicPartition -> {
+                    String t = topicPartition.topic();
+                    if (!cache.containsKey(t)) {
+                        cache.put(t, new HashSet<>());
+                    }
+                    cache.get(t).add(groupId);
+                });
+            });
+
+            topicSubscribedInfo.refresh(cache);
+        }
+
+        Set<String> groups = topicSubscribedInfo.getSubscribedGroups(topic);
+
+        Map<String, Object> res = new HashMap<>();
+        Collection<ConsumerConsole.TopicPartitionConsumeInfo> consumerDetail = consumerConsole.getConsumerDetail(groups);
+        List<ConsumerDetailVO> collect = consumerDetail.stream().filter(c -> topic.equals(c.topicPartition().topic())).map(ConsumerDetailVO::from).collect(Collectors.toList());
+        Map<String, List<ConsumerDetailVO>> map = collect.stream().collect(Collectors.groupingBy(ConsumerDetailVO::getGroupId));
+
+        map.forEach((groupId, list) -> {
+            Map<String, Object> sorting = new HashMap<>();
+            Collections.sort(list);
+            sorting.put("data", list);
+            sorting.put("lag", list.stream().map(ConsumerDetailVO::getLag).reduce(Long::sum));
+            res.put(groupId, sorting);
+        });
+
+        return ResponseData.create().data(res).success();
+    }
+
+    class TopicSubscribedInfo {
+        long lastTime = System.currentTimeMillis();
+
+        long refreshThreshold = 120 * 1000;
+
+        Map<String, Set<String>> cache = new HashMap<>();
+
+        public void refresh(Map<String, Set<String>> newCache) {
+            cache = newCache;
+            lastTime = System.currentTimeMillis();
+        }
+
+        public Set<String> getSubscribedGroups(String topic) {
+            return cache.getOrDefault(topic, Collections.emptySet());
+        }
+
+        public boolean isNeedRefresh(String topic) {
+            return System.currentTimeMillis() - lastTime > refreshThreshold || !cache.containsKey(topic);
+        }
+
+    }
 }

src/main/scala/kafka/console/ConsumerConsole.scala

@@ -173,6 +173,10 @@ class ConsumerConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaCon
         }).asInstanceOf[(Boolean, String)]
     }
 
+    /**
+     *
+     * @return k: topic, v: list[topic].
+     */
     def listSubscribeTopics(groupId: String): util.Map[String, util.List[TopicPartition]] = {
         val commitOffs = getCommittedOffsets(groupId)
         val map: util.Map[String, util.List[TopicPartition]] = new util.HashMap[String, util.List[TopicPartition]]()
@@ -185,6 +189,32 @@ class ConsumerConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaCon
         map
     }
 
+    /**
+     *
+     * @return k: groupId, v: list[topic].
+     */
+    def listSubscribeTopics(groups: util.Set[String]): util.Map[String, util.List[TopicPartition]] = {
+        val map: util.Map[String, util.List[TopicPartition]] = new util.HashMap[String, util.List[TopicPartition]]()
+        withAdminClientAndCatchError(admin => {
+            for(groupId <- groups.asScala) {
+                val commitOffs = admin.listConsumerGroupOffsets(
+                    groupId
+                ).partitionsToOffsetAndMetadata.get.asScala
+
+                for (t <- commitOffs.keySet) {
+                    if (!map.containsKey(groupId)) {
+                        map.put(groupId, new util.ArrayList[TopicPartition]())
+                    }
+                    map.get(groupId).add(t)
+                }
+            }
+            map
+        }, e => {
+            log.error("listSubscribeTopics error.", e)
+            map
+        }).asInstanceOf[util.Map[String, util.List[TopicPartition]]]
+    }
+
     private def describeConsumerGroups(groupIds: util.Set[String]): mutable.Map[String, ConsumerGroupDescription] = {
         withAdminClientAndCatchError(admin => {
             admin.describeConsumerGroups(groupIds).describedGroups().asScala.map {

ui/src/utils/api.js

@@ -116,6 +116,10 @@ export const KafkaConsumerApi = {
     url: "/consumer/topic/list",
     method: "get",
   },
+  getTopicSubscribedByGroups: {
+    url: "/consumer/topic/subscribed",
+    method: "get",
+  },
 };
 
 export const KafkaClusterApi = {

ui/src/views/topic/ConsumedDetail.vue

@@ -0,0 +1,190 @@
+<template>
+  <a-modal
+    :title="'Topic: ' + topic"
+    :visible="show"
+    :width="1200"
+    :mask="false"
+    :destroyOnClose="true"
+    :footer="null"
+    :maskClosable="false"
+    @cancel="handleCancel"
+  >
+    <div>
+      <a-spin :spinning="loading">
+        <div v-for="(v, k) in data" :key="k">
+          <strong>消费组: </strong><span class="color-font">{{ k }}</span
+          ><strong> | 积压: </strong><span class="color-font">{{ v.lag }}</span>
+          <hr />
+          <a-table
+            :columns="columns"
+            :data-source="v.data"
+            bordered
+            :rowKey="(record) => record.topic + record.partition"
+          >
+            <span slot="clientId" slot-scope="text, record">
+              <span v-if="text"> {{ text }}@{{ record.host }} </span>
+            </span>
+          </a-table>
+        </div>
+      </a-spin>
+    </div>
+  </a-modal>
+</template>
+
+<script>
+import request from "@/utils/request";
+import { KafkaConsumerApi } from "@/utils/api";
+import notification from "ant-design-vue/es/notification";
+
+export default {
+  name: "ConsumedDetail",
+  props: {
+    topic: {
+      type: String,
+      default: "",
+    },
+    visible: {
+      type: Boolean,
+      default: false,
+    },
+  },
+  data() {
+    return {
+      columns: columns,
+      show: this.visible,
+      data: [],
+      loading: false,
+      showResetPartitionOffsetDialog: false,
+      select: {
+        topic: "",
+        partition: 0,
+      },
+      resetPartitionOffsetForm: this.$form.createForm(this, {
+        name: "resetPartitionOffsetForm",
+      }),
+    };
+  },
+  watch: {
+    visible(v) {
+      this.show = v;
+      if (this.show) {
+        this.getConsumerDetail();
+      }
+    },
+  },
+  methods: {
+    getConsumerDetail() {
+      this.loading = true;
+      request({
+        url:
+          KafkaConsumerApi.getTopicSubscribedByGroups.url +
+          "?topic=" +
+          this.topic,
+        method: KafkaConsumerApi.getTopicSubscribedByGroups.method,
+      }).then((res) => {
+        this.loading = false;
+        if (res.code != 0) {
+          notification.error({
+            message: "error",
+            description: res.msg,
+          });
+        } else {
+          this.data = res.data;
+        }
+      });
+    },
+    handleCancel() {
+      this.data = [];
+      this.$emit("closeConsumedDetailDialog", {});
+    },
+    resetTopicOffsetToEndpoint(groupId, topic, type) {
+      this.requestResetOffset({
+        groupId: groupId,
+        topic: topic,
+        level: 1,
+        type: type,
+      });
+    },
+    requestResetOffset(data, callbackOnSuccess) {
+      this.loading = true;
+      request({
+        url: KafkaConsumerApi.resetOffset.url,
+        method: KafkaConsumerApi.resetOffset.method,
+        data: data,
+      }).then((res) => {
+        this.loading = false;
+        if (res.code != 0) {
+          notification.error({
+            message: "error",
+            description: res.msg,
+          });
+        } else {
+          this.$message.success(res.msg);
+          this.getConsumerDetail();
+          if (callbackOnSuccess) {
+            callbackOnSuccess();
+          }
+        }
+      });
+    },
+    openResetPartitionOffsetDialog(topic, partition) {
+      this.showResetPartitionOffsetDialog = true;
+      this.select.topic = topic;
+      this.select.partition = partition;
+    },
+    closeResetPartitionOffsetDialog() {
+      this.showResetPartitionOffsetDialog = false;
+    },
+    resetPartitionOffset() {
+      this.resetPartitionOffsetForm.validateFields((err, values) => {
+        if (!err) {
+          const data = Object.assign({}, values);
+          Object.assign(data, this.select);
+          data.groupId = this.group;
+          data.level = 2;
+          data.type = 4;
+          this.requestResetOffset(data, this.closeResetPartitionOffsetDialog());
+        }
+      });
+    },
+  },
+};
+
+const columns = [
+  {
+    title: "分区",
+    dataIndex: "partition",
+    key: "partition",
+  },
+  {
+    title: "客户端",
+    dataIndex: "clientId",
+    key: "clientId",
+    scopedSlots: { customRender: "clientId" },
+  },
+  {
+    title: "日志位点",
+    dataIndex: "logEndOffset",
+    key: "logEndOffset",
+  },
+  {
+    title: "消费位点",
+    dataIndex: "consumerOffset",
+    key: "consumerOffset",
+  },
+  {
+    title: "积压",
+    dataIndex: "lag",
+    key: "lag",
+  },
+];
+</script>
+
+<style scoped>
+.color-font {
+  color: dodgerblue;
+}
+#resetPartitionOffsetModal .ant-input-number {
+  width: 100% !important;
+}
+</style>

ui/src/views/topic/Topic.vue

@@ -77,6 +77,13 @@
               @click="openAddPartitionDialog(record.name)"
               >增加分区
             </a-button>
+            <a-button
+              size="small"
+              href="javascript:;"
+              class="operation-btn"
+              @click="openConsumedDetailDialog(record.name)"
+              >消费详情
+            </a-button>
           </div>
         </a-table>
         <PartitionInfo
@@ -94,6 +101,12 @@
           :topic="selectDetail.resourceName"
           @closeAddPartitionDialog="closeAddPartitionDialog"
         ></AddPartition>
+        <ConsumedDetail
+          :visible="showConsumedDetailDialog"
+          :topic="selectDetail.resourceName"
+          @closeConsumedDetailDialog="closeConsumedDetailDialog"
+        >
+        </ConsumedDetail>
       </div>
     </a-spin>
   </div>
@@ -106,10 +119,11 @@ import notification from "ant-design-vue/es/notification";
 import PartitionInfo from "@/views/topic/PartitionInfo";
 import CreateTopic from "@/views/topic/CreateTopic";
 import AddPartition from "@/views/topic/AddPartition";
+import ConsumedDetail from "@/views/topic/ConsumedDetail";
 
 export default {
   name: "Topic",
-  components: { PartitionInfo, CreateTopic, AddPartition },
+  components: { PartitionInfo, CreateTopic, AddPartition, ConsumedDetail },
   data() {
     return {
       queryParam: { type: "normal" },
@@ -128,6 +142,7 @@ export default {
       loading: false,
       showCreateTopic: false,
       showAddPartition: false,
+      showConsumedDetailDialog: false,
     };
   },
   methods: {
@@ -194,6 +209,13 @@ export default {
         this.getTopicList();
       }
     },
+    openConsumedDetailDialog(topic) {
+      this.showConsumedDetailDialog = true;
+      this.selectDetail.resourceName = topic;
+    },
+    closeConsumedDetailDialog() {
+      this.showConsumedDetailDialog = false;
+    },
   },
   created() {
     this.getTopicList();