Connect相关代码

2025-12-24 11:52:08 +08:00 · 2022-12-06 18:13:44 +08:00
parent fb5964af84
commit e8652d5db5
169 changed files with 6472 additions and 317 deletions
--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/AbstractMetricCollector.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/AbstractMetricCollector.java
@@ -1,52 +1,26 @@
 package com.xiaojukeji.know.streaming.km.collector.metric;

-import com.didiglobal.logi.log.ILog;
-import com.didiglobal.logi.log.LogFactory;
 import com.xiaojukeji.know.streaming.km.collector.service.CollectThreadPoolService;
-import com.xiaojukeji.know.streaming.km.common.utils.LoggerUtil;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
 import com.xiaojukeji.know.streaming.km.common.bean.event.metric.BaseMetricEvent;
 import com.xiaojukeji.know.streaming.km.common.component.SpringTool;
 import com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemTypeEnum;
-import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
 import com.xiaojukeji.know.streaming.km.common.utils.FutureWaitUtil;
 import org.springframework.beans.factory.annotation.Autowired;

-import java.util.List;


 /**
 * @author didi
 */
-public abstract class AbstractMetricCollector<T> {
-    protected static final ILog LOGGER                  = LogFactory.getLog(AbstractMetricCollector.class);
-
-    protected static final ILog METRIC_COLLECTED_LOGGER = LoggerUtil.getMetricCollectedLogger();
-
-    public abstract List<T> collectKafkaMetrics(ClusterPhy clusterPhy);
+public abstract class AbstractMetricCollector<M, C> {
+    public abstract String getClusterVersion(C c);

    public abstract VersionItemTypeEnum collectorType();

    @Autowired
    private CollectThreadPoolService collectThreadPoolService;

-    public void collectMetrics(ClusterPhy clusterPhy) {
-        long startTime = System.currentTimeMillis();
-
-        // 采集指标
-        List<T> metricsList = this.collectKafkaMetrics(clusterPhy);
-
-        // 输出耗时信息
-        LOGGER.info(
-                "metricType={}||clusterPhyId={}||costTimeUnitMs={}",
-                this.collectorType().getMessage(), clusterPhy.getId(), System.currentTimeMillis() - startTime
-        );
-
-        // 输出采集到的指标信息
-        METRIC_COLLECTED_LOGGER.debug("metricType={}||clusterPhyId={}||metrics={}!",
-                this.collectorType().getMessage(), clusterPhy.getId(), ConvertUtil.obj2Json(metricsList)
-        );
-    }
+    public abstract void collectMetrics(C c);

    protected FutureWaitUtil<Void> getFutureUtilByClusterPhyId(Long clusterPhyId) {
        return collectThreadPoolService.selectSuitableFutureUtil(clusterPhyId * 1000L + this.collectorType().getCode());
--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/connect/AbstractConnectMetricCollector.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/connect/AbstractConnectMetricCollector.java
@@ -0,0 +1,50 @@
+package com.xiaojukeji.know.streaming.km.collector.metric.connect;
+
+import com.didiglobal.logi.log.ILog;
+import com.didiglobal.logi.log.LogFactory;
+import com.xiaojukeji.know.streaming.km.collector.metric.AbstractMetricCollector;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.ConnectCluster;
+import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
+import com.xiaojukeji.know.streaming.km.common.utils.LoggerUtil;
+import com.xiaojukeji.know.streaming.km.core.service.connect.cluster.ConnectClusterService;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import java.util.List;
+
+/**
+ * @author didi
+ */
+public abstract class AbstractConnectMetricCollector<M> extends AbstractMetricCollector<M, ConnectCluster> {
+    private static final ILog LOGGER                    = LogFactory.getLog(AbstractConnectMetricCollector.class);
+
+    protected static final ILog METRIC_COLLECTED_LOGGER = LoggerUtil.getMetricCollectedLogger();
+
+    @Autowired
+    private ConnectClusterService connectClusterService;
+
+    public abstract List<M> collectConnectMetrics(ConnectCluster connectCluster);
+
+    @Override
+    public String getClusterVersion(ConnectCluster connectCluster){
+        return connectClusterService.getClusterVersion(connectCluster.getId());
+    }
+
+    @Override
+    public void collectMetrics(ConnectCluster connectCluster) {
+        long startTime = System.currentTimeMillis();
+
+        // 采集指标
+        List<M> metricsList = this.collectConnectMetrics(connectCluster);
+
+        // 输出耗时信息
+        LOGGER.info(
+                "metricType={}||connectClusterId={}||costTimeUnitMs={}",
+                this.collectorType().getMessage(), connectCluster.getId(), System.currentTimeMillis() - startTime
+        );
+
+        // 输出采集到的指标信息
+        METRIC_COLLECTED_LOGGER.debug("metricType={}||connectClusterId={}||metrics={}!",
+                this.collectorType().getMessage(), connectCluster.getId(), ConvertUtil.obj2Json(metricsList)
+        );
+    }
+}
--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/connect/ConnectClusterMetricCollector.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/connect/ConnectClusterMetricCollector.java
@@ -0,0 +1,83 @@
+package com.xiaojukeji.know.streaming.km.collector.metric.connect;
+
+import com.didiglobal.logi.log.ILog;
+import com.didiglobal.logi.log.LogFactory;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.ConnectCluster;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.connect.ConnectClusterMetrics;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.version.VersionControlItem;
+import com.xiaojukeji.know.streaming.km.common.bean.event.metric.connect.ConnectClusterMetricEvent;
+import com.xiaojukeji.know.streaming.km.common.constant.Constant;
+import com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemTypeEnum;
+import com.xiaojukeji.know.streaming.km.common.utils.FutureWaitUtil;
+import com.xiaojukeji.know.streaming.km.core.service.connect.cluster.ConnectClusterMetricService;
+import com.xiaojukeji.know.streaming.km.core.service.version.VersionControlService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.Collections;
+import java.util.List;
+
+import static com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemTypeEnum.METRIC_CONNECT_CLUSTER;
+
+/**
+ * @author didi
+ */
+@Component
+public class ConnectClusterMetricCollector extends AbstractConnectMetricCollector<ConnectClusterMetrics> {
+    protected static final ILog LOGGER = LogFactory.getLog(ConnectClusterMetricCollector.class);
+
+    @Autowired
+    private VersionControlService versionControlService;
+
+    @Autowired
+    private ConnectClusterMetricService connectClusterMetricService;
+
+    @Override
+    public List<ConnectClusterMetrics> collectConnectMetrics(ConnectCluster connectCluster) {
+        Long startTime          = System.currentTimeMillis();
+        Long clusterPhyId       = connectCluster.getKafkaClusterPhyId();
+        Long connectClusterId   = connectCluster.getId();
+
+        ConnectClusterMetrics metrics = new ConnectClusterMetrics(clusterPhyId, connectClusterId);
+        metrics.putMetric(Constant.COLLECT_METRICS_COST_TIME_METRICS_NAME, Constant.COLLECT_METRICS_ERROR_COST_TIME);
+        List<VersionControlItem> items = versionControlService.listVersionControlItem(getClusterVersion(connectCluster), collectorType().getCode());
+        FutureWaitUtil<Void> future = this.getFutureUtilByClusterPhyId(connectClusterId);
+
+        for (VersionControlItem item : items) {
+            future.runnableTask(
+                    String.format("class=ConnectClusterMetricCollector||connectClusterId=%d||metricName=%s", connectClusterId, item.getName()),
+                    30000,
+                    () -> {
+                        try {
+                            Result<ConnectClusterMetrics> ret = connectClusterMetricService.collectConnectClusterMetricsFromKafka(connectClusterId, item.getName());
+                            if (null == ret || !ret.hasData()) {
+                                return null;
+                            }
+                            metrics.putMetric(ret.getData().getMetrics());
+
+                        } catch (Exception e) {
+                            LOGGER.error(
+                                    "method=collectConnectMetrics||connectClusterId={}||metricName={}||errMsg=exception!",
+                                    connectClusterId, item.getName(), e
+                            );
+                        }
+                        return null;
+                    }
+            );
+        }
+
+        future.waitExecute(30000);
+
+        metrics.putMetric(Constant.COLLECT_METRICS_COST_TIME_METRICS_NAME, (System.currentTimeMillis() - startTime) / 1000.0f);
+
+        this.publishMetric(new ConnectClusterMetricEvent(this, Collections.singletonList(metrics)));
+
+        return Collections.singletonList(metrics);
+    }
+
+    @Override
+    public VersionItemTypeEnum collectorType() {
+        return METRIC_CONNECT_CLUSTER;
+    }
+}
--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/connect/ConnectConnectorMetricCollector.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/connect/ConnectConnectorMetricCollector.java
@@ -0,0 +1,102 @@
+package com.xiaojukeji.know.streaming.km.collector.metric.connect;
+
+import com.didiglobal.logi.log.ILog;
+import com.didiglobal.logi.log.LogFactory;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.ConnectCluster;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.connect.ConnectorMetrics;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.version.VersionControlItem;
+import com.xiaojukeji.know.streaming.km.common.bean.event.metric.connect.ConnectorMetricEvent;
+import com.xiaojukeji.know.streaming.km.common.constant.Constant;
+import com.xiaojukeji.know.streaming.km.common.enums.connect.ConnectorTypeEnum;
+import com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemTypeEnum;
+import com.xiaojukeji.know.streaming.km.common.utils.FutureWaitUtil;
+import com.xiaojukeji.know.streaming.km.core.service.connect.connector.ConnectorMetricService;
+import com.xiaojukeji.know.streaming.km.core.service.connect.connector.ConnectorService;
+import com.xiaojukeji.know.streaming.km.core.service.version.VersionControlService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemTypeEnum.METRIC_CONNECT_CONNECTOR;
+
+/**
+ * @author didi
+ */
+@Component
+public class ConnectConnectorMetricCollector extends AbstractConnectMetricCollector<ConnectorMetrics> {
+    protected static final ILog LOGGER = LogFactory.getLog(ConnectConnectorMetricCollector.class);
+
+    @Autowired
+    private VersionControlService versionControlService;
+
+    @Autowired
+    private ConnectorService connectorService;
+
+    @Autowired
+    private ConnectorMetricService connectorMetricService;
+
+    @Override
+    public List<ConnectorMetrics> collectConnectMetrics(ConnectCluster connectCluster) {
+        Long clusterPhyId       = connectCluster.getKafkaClusterPhyId();
+        Long connectClusterId   = connectCluster.getId();
+
+        List<VersionControlItem> items      = versionControlService.listVersionControlItem(this.getClusterVersion(connectCluster), collectorType().getCode());
+        Result<List<String>> connectorList  = connectorService.listConnectorsFromCluster(connectClusterId);
+
+        FutureWaitUtil<Void> future         = this.getFutureUtilByClusterPhyId(connectClusterId);
+
+        List<ConnectorMetrics> metricsList = new ArrayList<>();
+        for (String connectorName : connectorList.getData()) {
+            ConnectorMetrics metrics = new ConnectorMetrics(connectClusterId, connectorName);
+            metrics.setClusterPhyId(clusterPhyId);
+
+            metricsList.add(metrics);
+            future.runnableTask(
+                    String.format("class=ConnectConnectorMetricCollector||connectClusterId=%d||connectorName=%s", connectClusterId, connectorName),
+                    30000,
+                    () -> collectMetrics(connectClusterId, connectorName, metrics, items)
+            );
+        }
+        future.waitResult(30000);
+
+        this.publishMetric(new ConnectorMetricEvent(this, metricsList));
+
+        return metricsList;
+    }
+
+    @Override
+    public VersionItemTypeEnum collectorType() {
+        return METRIC_CONNECT_CONNECTOR;
+    }
+
+    /**************************************************** private method ****************************************************/
+
+    private void collectMetrics(Long connectClusterId, String connectorName, ConnectorMetrics metrics, List<VersionControlItem> items) {
+        long startTime = System.currentTimeMillis();
+        ConnectorTypeEnum connectorType = connectorService.getConnectorType(connectClusterId, connectorName);
+
+        metrics.putMetric(Constant.COLLECT_METRICS_COST_TIME_METRICS_NAME, Constant.COLLECT_METRICS_ERROR_COST_TIME);
+
+        for (VersionControlItem v : items) {
+            try {
+                Result<ConnectorMetrics> ret = connectorMetricService.collectConnectClusterMetricsFromKafka(connectClusterId, connectorName, v.getName(), connectorType);
+                if (null == ret || ret.failed() || null == ret.getData()) {
+                    continue;
+                }
+
+                metrics.putMetric(ret.getData().getMetrics());
+            } catch (Exception e) {
+                LOGGER.error(
+                        "method=collectMetrics||connectClusterId={}||connectorName={}||metric={}||errMsg=exception!",
+                        connectClusterId, connectorName, v.getName(), e
+                );
+            }
+        }
+
+        // 记录采集性能
+        metrics.putMetric(Constant.COLLECT_METRICS_COST_TIME_METRICS_NAME, (System.currentTimeMillis() - startTime) / 1000.0f);
+    }
+}
--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/kafka/AbstractKafkaMetricCollector.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/kafka/AbstractKafkaMetricCollector.java
@@ -0,0 +1,50 @@
+package com.xiaojukeji.know.streaming.km.collector.metric.kafka;
+
+import com.didiglobal.logi.log.ILog;
+import com.didiglobal.logi.log.LogFactory;
+import com.xiaojukeji.know.streaming.km.collector.metric.AbstractMetricCollector;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
+import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
+import com.xiaojukeji.know.streaming.km.common.utils.LoggerUtil;
+import com.xiaojukeji.know.streaming.km.core.service.cluster.ClusterPhyService;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import java.util.List;
+
+/**
+ * @author didi
+ */
+public abstract class AbstractKafkaMetricCollector<M> extends AbstractMetricCollector<M, ClusterPhy> {
+    private static final ILog LOGGER                    = LogFactory.getLog(AbstractMetricCollector.class);
+
+    protected static final ILog METRIC_COLLECTED_LOGGER = LoggerUtil.getMetricCollectedLogger();
+
+    @Autowired
+    private ClusterPhyService clusterPhyService;
+
+    public abstract List<M> collectKafkaMetrics(ClusterPhy clusterPhy);
+
+    @Override
+    public String getClusterVersion(ClusterPhy clusterPhy){
+        return clusterPhyService.getVersionFromCacheFirst(clusterPhy.getId());
+    }
+
+    @Override
+    public void collectMetrics(ClusterPhy clusterPhy) {
+        long startTime = System.currentTimeMillis();
+
+        // 采集指标
+        List<M> metricsList = this.collectKafkaMetrics(clusterPhy);
+
+        // 输出耗时信息
+        LOGGER.info(
+                "metricType={}||clusterPhyId={}||costTimeUnitMs={}",
+                this.collectorType().getMessage(), clusterPhy.getId(), System.currentTimeMillis() - startTime
+        );
+
+        // 输出采集到的指标信息
+        METRIC_COLLECTED_LOGGER.debug("metricType={}||clusterPhyId={}||metrics={}!",
+                this.collectorType().getMessage(), clusterPhy.getId(), ConvertUtil.obj2Json(metricsList)
+        );
+    }
+}
--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/kafka/BrokerMetricCollector.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/kafka/BrokerMetricCollector.java
@@ -2,7 +2,6 @@ package com.xiaojukeji.know.streaming.km.collector.metric.kafka;

 import com.didiglobal.logi.log.ILog;
 import com.didiglobal.logi.log.LogFactory;
-import com.xiaojukeji.know.streaming.km.collector.metric.AbstractMetricCollector;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.broker.Broker;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.BrokerMetrics;
@@ -27,7 +26,7 @@ import static com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemT
 * @author didi
 */
@Component
-public class BrokerMetricCollector extends AbstractMetricCollector<BrokerMetrics> {
+public class BrokerMetricCollector extends AbstractKafkaMetricCollector<BrokerMetrics> {
    private static final ILog LOGGER = LogFactory.getLog(BrokerMetricCollector.class);

    @Autowired
@@ -44,7 +43,7 @@ public class BrokerMetricCollector extends AbstractMetricCollector<BrokerMetrics
        Long clusterPhyId  = clusterPhy.getId();

        List<Broker>             brokers = brokerService.listAliveBrokersFromDB(clusterPhy.getId());
-        List<VersionControlItem> items   = versionControlService.listVersionControlItem(clusterPhyId, collectorType().getCode());
+        List<VersionControlItem> items   = versionControlService.listVersionControlItem(this.getClusterVersion(clusterPhy), collectorType().getCode());

        FutureWaitUtil<Void> future = this.getFutureUtilByClusterPhyId(clusterPhyId);

--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/kafka/ClusterMetricCollector.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/kafka/ClusterMetricCollector.java
@@ -2,7 +2,6 @@ package com.xiaojukeji.know.streaming.km.collector.metric.kafka;

 import com.didiglobal.logi.log.ILog;
 import com.didiglobal.logi.log.LogFactory;
-import com.xiaojukeji.know.streaming.km.collector.metric.AbstractMetricCollector;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.ClusterMetrics;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
@@ -25,7 +24,7 @@ import static com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemT
 * @author didi
 */
@Component
-public class ClusterMetricCollector extends AbstractMetricCollector<ClusterMetrics> {
+public class ClusterMetricCollector extends AbstractKafkaMetricCollector<ClusterMetrics> {
    protected static final ILog LOGGER = LogFactory.getLog(ClusterMetricCollector.class);

    @Autowired
@@ -38,7 +37,7 @@ public class ClusterMetricCollector extends AbstractMetricCollector<ClusterMetri
    public List<ClusterMetrics> collectKafkaMetrics(ClusterPhy clusterPhy) {
        Long        startTime           =   System.currentTimeMillis();
        Long        clusterPhyId        =   clusterPhy.getId();
-        List<VersionControlItem> items  =   versionControlService.listVersionControlItem(clusterPhyId, collectorType().getCode());
+        List<VersionControlItem> items  =   versionControlService.listVersionControlItem(this.getClusterVersion(clusterPhy), collectorType().getCode());

        ClusterMetrics metrics = new ClusterMetrics(clusterPhyId, clusterPhy.getKafkaVersion());
        metrics.putMetric(Constant.COLLECT_METRICS_COST_TIME_METRICS_NAME, Constant.COLLECT_METRICS_ERROR_COST_TIME);
--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/kafka/GroupMetricCollector.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/kafka/GroupMetricCollector.java
@@ -2,7 +2,6 @@ package com.xiaojukeji.know.streaming.km.collector.metric.kafka;

 import com.didiglobal.logi.log.ILog;
 import com.didiglobal.logi.log.LogFactory;
-import com.xiaojukeji.know.streaming.km.collector.metric.AbstractMetricCollector;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.GroupMetrics;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
@@ -28,7 +27,7 @@ import static com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemT
 * @author didi
 */
@Component
-public class GroupMetricCollector extends AbstractMetricCollector<GroupMetrics> {
+public class GroupMetricCollector extends AbstractKafkaMetricCollector<GroupMetrics> {
    protected static final ILog  LOGGER = LogFactory.getLog(GroupMetricCollector.class);

    @Autowired
@@ -46,7 +45,7 @@ public class GroupMetricCollector extends AbstractMetricCollector<GroupMetrics>

        List<String> groupNameList = new ArrayList<>();
        try {
-            groupNameList = groupService.listGroupsFromKafka(clusterPhyId);
+            groupNameList = groupService.listGroupsFromKafka(clusterPhy);
        } catch (Exception e) {
            LOGGER.error("method=collectKafkaMetrics||clusterPhyId={}||msg=exception!", clusterPhyId, e);
        }
@@ -55,7 +54,7 @@ public class GroupMetricCollector extends AbstractMetricCollector<GroupMetrics>
            return Collections.emptyList();
        }

-        List<VersionControlItem> items = versionControlService.listVersionControlItem(clusterPhyId, collectorType().getCode());
+        List<VersionControlItem> items = versionControlService.listVersionControlItem(this.getClusterVersion(clusterPhy), collectorType().getCode());

        FutureWaitUtil<Void> future = this.getFutureUtilByClusterPhyId(clusterPhyId);

--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/kafka/PartitionMetricCollector.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/kafka/PartitionMetricCollector.java
@@ -2,7 +2,6 @@ package com.xiaojukeji.know.streaming.km.collector.metric.kafka;

 import com.didiglobal.logi.log.ILog;
 import com.didiglobal.logi.log.LogFactory;
-import com.xiaojukeji.know.streaming.km.collector.metric.AbstractMetricCollector;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.PartitionMetrics;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
@@ -26,7 +25,7 @@ import static com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemT
 * @author didi
 */
@Component
-public class PartitionMetricCollector extends AbstractMetricCollector<PartitionMetrics> {
+public class PartitionMetricCollector extends AbstractKafkaMetricCollector<PartitionMetrics> {
    protected static final ILog  LOGGER = LogFactory.getLog(PartitionMetricCollector.class);

    @Autowired
@@ -42,7 +41,7 @@ public class PartitionMetricCollector extends AbstractMetricCollector<PartitionM
    public List<PartitionMetrics> collectKafkaMetrics(ClusterPhy clusterPhy) {
        Long        clusterPhyId        =   clusterPhy.getId();
        List<Topic> topicList           =   topicService.listTopicsFromCacheFirst(clusterPhyId);
-        List<VersionControlItem> items  =   versionControlService.listVersionControlItem(clusterPhyId, collectorType().getCode());
+        List<VersionControlItem> items  =   versionControlService.listVersionControlItem(this.getClusterVersion(clusterPhy), collectorType().getCode());

        FutureWaitUtil<Void> future = this.getFutureUtilByClusterPhyId(clusterPhyId);

--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/kafka/ReplicaMetricCollector.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/kafka/ReplicaMetricCollector.java
@@ -2,7 +2,6 @@ package com.xiaojukeji.know.streaming.km.collector.metric.kafka;

 import com.didiglobal.logi.log.ILog;
 import com.didiglobal.logi.log.LogFactory;
-import com.xiaojukeji.know.streaming.km.collector.metric.AbstractMetricCollector;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.ReplicationMetrics;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.partition.Partition;
@@ -27,7 +26,7 @@ import static com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemT
 * @author didi
 */
@Component
-public class ReplicaMetricCollector extends AbstractMetricCollector<ReplicationMetrics> {
+public class ReplicaMetricCollector extends AbstractKafkaMetricCollector<ReplicationMetrics> {
    protected static final ILog  LOGGER = LogFactory.getLog(ReplicaMetricCollector.class);

    @Autowired
@@ -42,8 +41,8 @@ public class ReplicaMetricCollector extends AbstractMetricCollector<ReplicationM
    @Override
    public List<ReplicationMetrics> collectKafkaMetrics(ClusterPhy clusterPhy) {
        Long        clusterPhyId        =   clusterPhy.getId();
-        List<VersionControlItem> items  =   versionControlService.listVersionControlItem(clusterPhyId, collectorType().getCode());
        List<Partition> partitions      =   partitionService.listPartitionFromCacheFirst(clusterPhyId);
+        List<VersionControlItem> items  =   versionControlService.listVersionControlItem(this.getClusterVersion(clusterPhy), collectorType().getCode());

        FutureWaitUtil<Void> future = this.getFutureUtilByClusterPhyId(clusterPhyId);

--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/kafka/TopicMetricCollector.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/kafka/TopicMetricCollector.java
@@ -2,7 +2,6 @@ package com.xiaojukeji.know.streaming.km.collector.metric.kafka;

 import com.didiglobal.logi.log.ILog;
 import com.didiglobal.logi.log.LogFactory;
-import com.xiaojukeji.know.streaming.km.collector.metric.AbstractMetricCollector;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.TopicMetrics;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
@@ -30,7 +29,7 @@ import static com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemT
 * @author didi
 */
@Component
-public class TopicMetricCollector extends AbstractMetricCollector<TopicMetrics> {
+public class TopicMetricCollector extends AbstractKafkaMetricCollector<TopicMetrics> {
    protected static final ILog  LOGGER = LogFactory.getLog(TopicMetricCollector.class);

    @Autowired
@@ -48,7 +47,7 @@ public class TopicMetricCollector extends AbstractMetricCollector<TopicMetrics>
    public List<TopicMetrics> collectKafkaMetrics(ClusterPhy clusterPhy) {
        Long        clusterPhyId        =   clusterPhy.getId();
        List<Topic> topics              =   topicService.listTopicsFromCacheFirst(clusterPhyId);
-        List<VersionControlItem> items  =   versionControlService.listVersionControlItem(clusterPhyId, collectorType().getCode());
+        List<VersionControlItem> items  =   versionControlService.listVersionControlItem(this.getClusterVersion(clusterPhy), collectorType().getCode());

        FutureWaitUtil<Void> future = this.getFutureUtilByClusterPhyId(clusterPhyId);

--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/kafka/ZookeeperMetricCollector.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/metric/kafka/ZookeeperMetricCollector.java
@@ -2,7 +2,6 @@ package com.xiaojukeji.know.streaming.km.collector.metric.kafka;

 import com.didiglobal.logi.log.ILog;
 import com.didiglobal.logi.log.LogFactory;
-import com.xiaojukeji.know.streaming.km.collector.metric.AbstractMetricCollector;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.config.ZKConfig;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.kafkacontroller.KafkaController;
@@ -34,7 +33,7 @@ import static com.xiaojukeji.know.streaming.km.common.enums.version.VersionItemT
 * @author didi
 */
@Component
-public class ZookeeperMetricCollector extends AbstractMetricCollector<ZookeeperMetrics> {
+public class ZookeeperMetricCollector extends AbstractKafkaMetricCollector<ZookeeperMetrics> {
    protected static final ILog  LOGGER = LogFactory.getLog(ZookeeperMetricCollector.class);

    @Autowired
@@ -53,7 +52,7 @@ public class ZookeeperMetricCollector extends AbstractMetricCollector<ZookeeperM
    public List<ZookeeperMetrics> collectKafkaMetrics(ClusterPhy clusterPhy) {
        Long        startTime           =   System.currentTimeMillis();
        Long        clusterPhyId        =   clusterPhy.getId();
-        List<VersionControlItem> items  =   versionControlService.listVersionControlItem(clusterPhyId, collectorType().getCode());
+        List<VersionControlItem> items  =   versionControlService.listVersionControlItem(this.getClusterVersion(clusterPhy), collectorType().getCode());
        List<ZookeeperInfo> aliveZKList =   zookeeperService.listFromDBByCluster(clusterPhyId)
                .stream()
                .filter(elem -> Constant.ALIVE.equals(elem.getStatus()))
--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/connect/ConnectClusterMetricESSender.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/connect/ConnectClusterMetricESSender.java
@@ -0,0 +1,33 @@
+package com.xiaojukeji.know.streaming.km.collector.sink.connect;
+
+import com.didiglobal.logi.log.ILog;
+import com.didiglobal.logi.log.LogFactory;
+import com.xiaojukeji.know.streaming.km.collector.sink.AbstractMetricESSender;
+import com.xiaojukeji.know.streaming.km.common.bean.event.metric.connect.ConnectClusterMetricEvent;
+import com.xiaojukeji.know.streaming.km.common.bean.po.metrice.connect.ConnectClusterMetricPO;
+import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
+import org.springframework.context.ApplicationListener;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+
+import static com.xiaojukeji.know.streaming.km.persistence.es.template.TemplateConstant.CONNECT_CLUSTER_INDEX;
+
+/**
+ * @author wyb
+ * @date 2022/11/7
+ */
+@Component
+public class ConnectClusterMetricESSender extends AbstractMetricESSender implements ApplicationListener<ConnectClusterMetricEvent> {
+    protected static final ILog LOGGER = LogFactory.getLog(ConnectClusterMetricESSender.class);
+
+    @PostConstruct
+    public void init(){
+        LOGGER.info("class=ConnectClusterMetricESSender||method=init||msg=init finished");
+    }
+
+    @Override
+    public void onApplicationEvent(ConnectClusterMetricEvent event) {
+        send2es(CONNECT_CLUSTER_INDEX, ConvertUtil.list2List(event.getConnectClusterMetrics(), ConnectClusterMetricPO.class));
+    }
+}
--- a/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/connect/ConnectorMetricESSender.java
+++ b/km-collector/src/main/java/com/xiaojukeji/know/streaming/km/collector/sink/connect/ConnectorMetricESSender.java
@@ -0,0 +1,33 @@
+package com.xiaojukeji.know.streaming.km.collector.sink.connect;
+
+import com.didiglobal.logi.log.ILog;
+import com.didiglobal.logi.log.LogFactory;
+import com.xiaojukeji.know.streaming.km.collector.sink.AbstractMetricESSender;
+import com.xiaojukeji.know.streaming.km.common.bean.event.metric.connect.ConnectorMetricEvent;
+import com.xiaojukeji.know.streaming.km.common.bean.po.metrice.connect.ConnectorMetricPO;
+import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
+import org.springframework.context.ApplicationListener;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+
+import static com.xiaojukeji.know.streaming.km.persistence.es.template.TemplateConstant.CONNECT_CONNECTOR_INDEX;
+
+/**
+ * @author wyb
+ * @date 2022/11/7
+ */
+@Component
+public class ConnectorMetricESSender extends AbstractMetricESSender implements ApplicationListener<ConnectorMetricEvent> {
+    protected static final ILog LOGGER = LogFactory.getLog(ConnectorMetricESSender.class);
+
+    @PostConstruct
+    public void init(){
+        LOGGER.info("class=ConnectorMetricESSender||method=init||msg=init finished");
+    }
+
+    @Override
+    public void onApplicationEvent(ConnectorMetricEvent event) {
+        send2es(CONNECT_CONNECTOR_INDEX, ConvertUtil.list2List(event.getConnectorMetricsList(), ConnectorMetricPO.class));
+    }
+}