查看消费端成员

src/main/java/com/xuxd/kafka/console/beans/TopicPartition.java

@@ -0,0 +1,29 @@
+package com.xuxd.kafka.console.beans;
+
+import lombok.Data;
+
+/**
+ * kafka-console-ui.
+ *
+ * @author xuxd
+ * @date 2021-09-26 20:32:15
+ **/
+@Data
+public class TopicPartition implements Comparable {
+
+    private final String topic;
+
+    private final int partition;
+
+    @Override public int compareTo(Object o) {
+        if (o == null) {
+            return -1;
+        }
+        TopicPartition other = (TopicPartition) o;
+        if (!this.topic.equals(other.getTopic())) {
+            return this.compareTo(other);
+        }
+
+        return this.partition - other.partition;
+    }
+}

src/main/java/com/xuxd/kafka/console/beans/vo/ConsumerMemberVO.java

@@ -0,0 +1,38 @@
+package com.xuxd.kafka.console.beans.vo;
+
+import com.xuxd.kafka.console.beans.TopicPartition;
+import java.util.Comparator;
+import java.util.List;
+import java.util.stream.Collectors;
+import lombok.Data;
+import org.apache.kafka.clients.admin.MemberDescription;
+
+/**
+ * kafka-console-ui.
+ *
+ * @author xuxd
+ * @date 2021-09-26 20:02:59
+ **/
+@Data
+public class ConsumerMemberVO {
+
+    private String memberId;
+    private String groupInstanceId;
+    private String clientId;
+    private String host;
+    private List<TopicPartition> partitions;
+
+    public static ConsumerMemberVO from(MemberDescription description) {
+        ConsumerMemberVO vo = new ConsumerMemberVO();
+        vo.setMemberId(description.consumerId());
+        vo.setGroupInstanceId(description.groupInstanceId().orElse(""));
+        vo.setClientId(description.clientId());
+        vo.setHost(description.host());
+        List<TopicPartition> collect = description.assignment().topicPartitions().stream().map(t -> new TopicPartition(t.topic(), t.partition())).collect(Collectors.toList());
+        collect.sort(Comparator.naturalOrder());
+        vo.setPartitions(collect);
+
+        return vo;
+    }
+
+}

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

@@ -12,6 +12,7 @@ import org.apache.commons.lang3.StringUtils;
 import org.apache.kafka.common.ConsumerGroupState;
 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;
@@ -49,4 +50,9 @@ public class ConsumerController {
     public Object deleteConsumerGroup(@RequestParam String groupId) {
         return consumerService.deleteConsumerGroup(groupId);
     }
+
+    @GetMapping("/member")
+    public Object getConsumerMembers(@RequestParam String groupId) {
+        return consumerService.getConsumerMembers(groupId);
+    }
 }

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

@@ -16,4 +16,6 @@ public interface ConsumerService {
     ResponseData getConsumerGroupList(List<String> groupIds, Set<ConsumerGroupState> states);
 
     ResponseData deleteConsumerGroup(String groupId);
+
+    ResponseData getConsumerMembers(String groupId);
 }

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

@@ -3,7 +3,9 @@ package com.xuxd.kafka.console.service.impl;
 import com.xuxd.kafka.console.beans.CounterList;
 import com.xuxd.kafka.console.beans.ResponseData;
 import com.xuxd.kafka.console.beans.vo.ConsumerGroupVO;
+import com.xuxd.kafka.console.beans.vo.ConsumerMemberVO;
 import com.xuxd.kafka.console.service.ConsumerService;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashSet;
@@ -11,6 +13,9 @@ import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
 import kafka.console.ConsumerConsole;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.kafka.clients.admin.ConsumerGroupDescription;
+import org.apache.kafka.clients.admin.MemberDescription;
 import org.apache.kafka.common.ConsumerGroupState;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
@@ -29,6 +34,7 @@ public class ConsumerServiceImpl implements ConsumerService {
     private ConsumerConsole consumerConsole;
 
     @Override public ResponseData getConsumerGroupList(List<String> groupIds, Set<ConsumerGroupState> states) {
+        String simulateGroup = "inner_xxx_not_exit_group_###";
         Set<String> groupList = new HashSet<>();
         if (groupIds != null && !groupIds.isEmpty()) {
             if (states != null && !states.isEmpty()) {
@@ -44,8 +50,15 @@ public class ConsumerServiceImpl implements ConsumerService {
             }
         } else {
             groupList.addAll(consumerConsole.getConsumerGroupIdList(states));
+            if (groupList.isEmpty()) {
+                // The consumer groupId that match the specified states could not find, so simulate an impossible groupId.
+                groupList.add(simulateGroup);
+            }
         }
         List<ConsumerGroupVO> consumerGroupVOS = consumerConsole.getConsumerGroupList(groupList).stream().map(c -> ConsumerGroupVO.from(c)).collect(Collectors.toList());
+        if (consumerGroupVOS.size() == 1 && consumerGroupVOS.get(0).getGroupId().equals(simulateGroup)) {
+            consumerGroupVOS.clear();
+        }
         consumerGroupVOS.sort(Comparator.comparing(ConsumerGroupVO::getGroupId));
         return ResponseData.create().data(new CounterList<>(consumerGroupVOS)).success();
     }
@@ -54,4 +67,15 @@ public class ConsumerServiceImpl implements ConsumerService {
         Tuple2<Object, String> tuple2 = consumerConsole.deleteConsumerGroups(Collections.singletonList(groupId));
         return (Boolean) tuple2._1 ? ResponseData.create().success() : ResponseData.create().failed(tuple2._2);
     }
+
+    @Override public ResponseData getConsumerMembers(String groupId) {
+        Set<ConsumerGroupDescription> groupList = consumerConsole.getConsumerGroupList(Collections.singleton(groupId));
+        if (CollectionUtils.isEmpty(groupList)) {
+            return ResponseData.create().data(Collections.emptyList()).success();
+        }
+        Collection<MemberDescription> members = groupList.stream().findFirst().get().members();
+        List<ConsumerMemberVO> vos = members.stream().map(ConsumerMemberVO::from).collect(Collectors.toList());
+        vos.sort(Comparator.comparing(ConsumerMemberVO::getClientId));
+        return ResponseData.create().data(vos).success();
+    }
 }

src/main/resources/application.yml

@@ -7,7 +7,7 @@ server:
 kafka:
   config:
     # kafka broker地址，多个以逗号分隔
-    bootstrap-server: 'localhost:9092'
+    bootstrap-server: '10.100.64.48:9092,10.100.77.250:9092,10.100.73.154:9092'
     request-timeout-ms: 60000
     # 服务端是否启用acl，如果不启用，下面的几项都忽略即可
     enable-acl: true
@@ -17,11 +17,11 @@ kafka:
     # 超级管理员用户名，在broker上已经配置为超级管理员
     admin-username: admin
     # 超级管理员密码
-    admin-password: admin
+    admin-password: admin!QAZ
     # 启动自动创建配置的超级管理员用户
     admin-create: false
     # broker连接的zk地址
-    zookeeper-addr: 'localhost:2181'
+    zookeeper-addr: '10.100.64.48:5181,10.100.77.250:5181,10.100.73.154:5181'
     sasl-jaas-config: org.apache.kafka.common.security.scram.ScramLoginModule required username="${kafka.config.admin-username}" password="${kafka.config.admin-password}";
 
 spring:

ui/src/utils/api.js

@@ -80,4 +80,8 @@ export const KafkaConsumerApi = {
     url: "/consumer/group",
     method: "delete",
   },
+  getConsumerMembers: {
+    url: "/consumer/member",
+    method: "get",
+  },
 };

ui/src/views/group/Group.vue

@@ -69,8 +69,10 @@
           bordered
           row-key="groupId"
         >
-          <div slot="members" slot-scope="text">
+          <div slot="members" slot-scope="text, record">
-            <a href="#">{{ text }} </a>
+            <a href="#" @click="openConsumerMemberDialog(record.groupId)"
+              >{{ text }}
+            </a>
           </div>
 
           <div slot="state" slot-scope="text">
@@ -91,6 +93,11 @@
             </a-popconfirm>
           </div>
         </a-table>
+        <Member
+          :visible="showConsumerGroupDialog"
+          :group="selectDetail.resourceName"
+          @closeConsumerMemberDialog="closeConsumerDialog"
+        ></Member>
       </div>
     </a-spin>
   </div>
@@ -100,10 +107,11 @@
 import request from "@/utils/request";
 import { KafkaConsumerApi } from "@/utils/api";
 import notification from "ant-design-vue/es/notification";
+import Member from "@/views/group/Member";
 
 export default {
   name: "ConsumerGroup",
-  components: {},
+  components: { Member },
   data() {
     return {
       queryParam: {},
@@ -121,6 +129,7 @@ export default {
         username: "",
       },
       loading: false,
+      showConsumerGroupDialog: false,
     };
   },
   methods: {
@@ -163,6 +172,13 @@ export default {
         }
       });
     },
+    openConsumerMemberDialog(groupId) {
+      this.showConsumerGroupDialog = true;
+      this.selectDetail.resourceName = groupId;
+    },
+    closeConsumerDialog() {
+      this.showConsumerGroupDialog = false;
+    },
   },
   created() {
     this.getConsumerGroupList();

ui/src/views/group/Member.vue

@@ -0,0 +1,124 @@
+<template>
+  <a-modal
+    title="消费端成员"
+    :visible="show"
+    :width="1800"
+    :mask="false"
+    :destroyOnClose="true"
+    :footer="null"
+    :maskClosable="false"
+    @cancel="handleCancel"
+  >
+    <div>
+      <a-spin :spinning="loading">
+        <a-table
+          :columns="columns"
+          :data-source="data"
+          bordered
+          :rowKey="(record) => record.memberId"
+        >
+          <ul slot="partitions" slot-scope="text">
+            <ol v-for="i in text" :key="i.topic + i.partition">
+              {{
+                i.topic
+              }}:
+              {{
+                i.partition
+              }}
+            </ol>
+          </ul>
+        </a-table>
+      </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: "Member",
+  props: {
+    group: {
+      type: String,
+      default: "",
+    },
+    visible: {
+      type: Boolean,
+      default: false,
+    },
+  },
+  data() {
+    return {
+      columns: columns,
+      show: this.visible,
+      data: [],
+      loading: false,
+    };
+  },
+  watch: {
+    visible(v) {
+      this.show = v;
+      if (this.show) {
+        this.getPartitionInfo();
+      }
+    },
+  },
+  methods: {
+    getPartitionInfo() {
+      this.loading = true;
+      request({
+        url: KafkaConsumerApi.getConsumerMembers.url + "?groupId=" + this.group,
+        method: KafkaConsumerApi.getConsumerMembers.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("closeConsumerMemberDialog", {});
+    },
+  },
+};
+
+const columns = [
+  {
+    title: "成员ID",
+    dataIndex: "memberId",
+    key: "memberId",
+  },
+  {
+    title: "客户端ID",
+    dataIndex: "clientId",
+    key: "clientId",
+  },
+  {
+    title: "实例ID",
+    dataIndex: "groupInstanceId",
+    key: "groupInstanceId",
+  },
+  {
+    title: "主机",
+    dataIndex: "host",
+    key: "host",
+  },
+  {
+    title: "分区信息",
+    dataIndex: "partitions",
+    key: "partitions",
+    scopedSlots: { customRender: "partitions" },
+  },
+];
+</script>
+
+<style scoped></style>