主页展示Broker API的版本兼容信息

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

@@ -0,0 +1,24 @@
+package com.xuxd.kafka.console.beans.vo;
+
+import java.util.List;
+import lombok.Data;
+
+/**
+ * kafka-console-ui.
+ *
+ * @author xuxd
+ * @date 2022-01-22 16:24:58
+ **/
+@Data
+public class BrokerApiVersionVO {
+
+    private int brokerId;
+
+    private String host;
+
+    private int supportNums;
+
+    private int unSupportNums;
+
+    private List<String> versionInfo;
+}

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

@@ -53,4 +53,9 @@ public class ClusterController {
     public Object peekClusterInfo() {
         return clusterService.peekClusterInfo();
     }
+
+    @GetMapping("/info/api/version")
+    public Object getBrokerApiVersionInfo() {
+        return clusterService.getBrokerApiVersionInfo();
+    }
 }

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

@@ -21,4 +21,6 @@ public interface ClusterService {
     ResponseData updateClusterInfo(ClusterInfoDO infoDO);
 
     ResponseData peekClusterInfo();
+
+    ResponseData getBrokerApiVersionInfo();
 }

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

@@ -3,13 +3,22 @@ package com.xuxd.kafka.console.service.impl;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.xuxd.kafka.console.beans.ResponseData;
 import com.xuxd.kafka.console.beans.dos.ClusterInfoDO;
+import com.xuxd.kafka.console.beans.vo.BrokerApiVersionVO;
 import com.xuxd.kafka.console.beans.vo.ClusterInfoVO;
 import com.xuxd.kafka.console.dao.ClusterInfoMapper;
 import com.xuxd.kafka.console.service.ClusterService;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
 import java.util.List;
 import java.util.stream.Collectors;
 import kafka.console.ClusterConsole;
 import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.kafka.clients.NodeApiVersions;
+import org.apache.kafka.common.Node;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.stereotype.Service;
 
@@ -69,4 +78,29 @@ public class ClusterServiceImpl implements ClusterService {
         return ResponseData.create().data(dos.stream().findFirst().map(ClusterInfoVO::from)).success();
     }
 
+    @Override public ResponseData getBrokerApiVersionInfo() {
+        HashMap<Node, NodeApiVersions> map = clusterConsole.listBrokerVersionInfo();
+        List<BrokerApiVersionVO> list = new ArrayList<>(map.size());
+        map.forEach(((node, versions) -> {
+            BrokerApiVersionVO vo = new BrokerApiVersionVO();
+            vo.setBrokerId(node.id());
+            vo.setHost(node.host() + ":" + node.port());
+            vo.setSupportNums(versions.allSupportedApiVersions().size());
+            String versionInfo = versions.toString(true);
+            int from = 0;
+            int count = 0;
+            int index = -1;
+            while ((index = versionInfo.indexOf("UNSUPPORTED", from)) >= 0 && from < versionInfo.length()) {
+                count++;
+                from = index + 1;
+            }
+            vo.setUnSupportNums(count);
+            versionInfo = versionInfo.substring(1, versionInfo.length() - 2);
+            vo.setVersionInfo(Arrays.asList(StringUtils.split(versionInfo, ",")));
+            list.add(vo);
+        }));
+        Collections.sort(list, Comparator.comparingInt(BrokerApiVersionVO::getBrokerId));
+        return ResponseData.create().data(list).success();
+    }
+
 }