8x48/KnowStreaming
1
0
Fork 0
mirror of https://github.com/didi/KnowStreaming.git synced 2025-12-24 11:52:08 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #473 from didi/dev

Browse Source 
Dev
This commit is contained in:
EricZeng
2022-03-07 14:49:52 +08:00
committed by GitHub
parent 482a375e31 44d537f78c
commit e51dbe0ca7
147 changed files with 1828 additions and 924 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
kafka-manager-core/pom.xml
View File

@@ -24,7 +24,6 @@
<java_target_version>1.8</java_target_version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<file_encoding>UTF-8</file_encoding>
<spring-version>5.1.3.RELEASE</spring-version>
</properties>
<dependencies>
@@ -38,12 +37,10 @@
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-web</artifactId>
<version>${spring-version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>${spring-version}</version>
</dependency>
<!-- javax -->

27
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/cache/KafkaClientPool.java vendored
View File

@@ -14,6 +14,8 @@ import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import java.util.Map;
import java.util.Properties;
@@ -25,9 +27,22 @@ import java.util.concurrent.locks.ReentrantLock;
* @author zengqiao
* @date 19/12/24
*/
@Service
public class KafkaClientPool {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaClientPool.class);
@Value(value = "${client-pool.kafka-consumer.min-idle-client-num:24}")
private Integer kafkaConsumerMinIdleClientNum;
@Value(value = "${client-pool.kafka-consumer.max-idle-client-num:24}")
private Integer kafkaConsumerMaxIdleClientNum;
@Value(value = "${client-pool.kafka-consumer.max-total-client-num:24}")
private Integer kafkaConsumerMaxTotalClientNum;
@Value(value = "${client-pool.kafka-consumer.borrow-timeout-unit-ms:3000}")
private Integer kafkaConsumerBorrowTimeoutUnitMs;
/**
* AdminClient
*/
@@ -84,7 +99,7 @@ public class KafkaClientPool {
return true;
}
private static void initKafkaConsumerPool(ClusterDO clusterDO) {
private void initKafkaConsumerPool(ClusterDO clusterDO) {
lock.lock();
try {
GenericObjectPool<KafkaConsumer<String, String>> objectPool = KAFKA_CONSUMER_POOL.get(clusterDO.getId());
@@ -92,9 +107,9 @@ public class KafkaClientPool {
return;
}
GenericObjectPoolConfig<KafkaConsumer<String, String>> config = new GenericObjectPoolConfig<>();
config.setMaxIdle(24);
config.setMinIdle(24);
config.setMaxTotal(24);
config.setMaxIdle(kafkaConsumerMaxIdleClientNum);
config.setMinIdle(kafkaConsumerMinIdleClientNum);
config.setMaxTotal(kafkaConsumerMaxTotalClientNum);
KAFKA_CONSUMER_POOL.put(clusterDO.getId(), new GenericObjectPool<>(new KafkaConsumerFactory(clusterDO), config));
} catch (Exception e) {
LOGGER.error("create kafka consumer pool failed, clusterDO:{}.", clusterDO, e);
@@ -118,7 +133,7 @@ public class KafkaClientPool {
}
}
public static KafkaConsumer<String, String> borrowKafkaConsumerClient(ClusterDO clusterDO) {
public KafkaConsumer<String, String> borrowKafkaConsumerClient(ClusterDO clusterDO) {
if (ValidateUtils.isNull(clusterDO)) {
return null;
}
@@ -132,7 +147,7 @@ public class KafkaClientPool {
}
try {
return objectPool.borrowObject(3000);
return objectPool.borrowObject(kafkaConsumerBorrowTimeoutUnitMs);
} catch (Exception e) {
LOGGER.error("borrow kafka consumer client failed, clusterDO:{}.", clusterDO, e);
}

6
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/cache/LogicalClusterMetadataManager.java vendored
View File

@@ -156,6 +156,9 @@ public class LogicalClusterMetadataManager {
return logicalClusterDO.getClusterId();
}
/**
* 定时刷新逻辑集群元数据到缓存中
*/
@Scheduled(cron="0/30 * * * * ?")
public void flush() {
List<LogicalClusterDO> logicalClusterDOList = logicalClusterService.listAll();
@@ -208,7 +211,8 @@ public class LogicalClusterMetadataManager {
// 计算逻辑集群到Topic名称的映射
Set<String> topicNameSet = PhysicalClusterMetadataManager.getBrokerTopicNum(
logicalClusterDO.getClusterId(),
brokerIdSet);
brokerIdSet
);
LOGICAL_CLUSTER_ID_TOPIC_NAME_MAP.put(logicalClusterDO.getId(), topicNameSet);
// 计算Topic名称到逻辑集群的映射

14
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/cache/PhysicalClusterMetadataManager.java vendored
View File

@@ -50,6 +50,9 @@ public class PhysicalClusterMetadataManager {
@Autowired
private ClusterService clusterService;
@Autowired
private ThreadPool threadPool;
private static final Map<Long, ClusterDO> CLUSTER_MAP = new ConcurrentHashMap<>();
private static final Map<Long, ControllerData> CONTROLLER_DATA_MAP = new ConcurrentHashMap<>();
@@ -125,7 +128,7 @@ public class PhysicalClusterMetadataManager {
zkConfig.watchChildren(ZkPathUtil.BROKER_IDS_ROOT, brokerListener);
//增加Topic监控
TopicStateListener topicListener = new TopicStateListener(clusterDO.getId(), zkConfig);
TopicStateListener topicListener = new TopicStateListener(clusterDO.getId(), zkConfig, threadPool);
topicListener.init();
zkConfig.watchChildren(ZkPathUtil.BROKER_TOPICS_ROOT, topicListener);
@@ -314,7 +317,7 @@ public class PhysicalClusterMetadataManager {
metadataMap.put(brokerId, brokerMetadata);
Map<Integer, JmxConnectorWrap> jmxMap = JMX_CONNECTOR_MAP.getOrDefault(clusterId, new ConcurrentHashMap<>());
jmxMap.put(brokerId, new JmxConnectorWrap(brokerMetadata.getHost(), brokerMetadata.getJmxPort(), jmxConfig));
jmxMap.put(brokerId, new JmxConnectorWrap(clusterId, brokerId, brokerMetadata.getHost(), brokerMetadata.getJmxPort(), jmxConfig));
JMX_CONNECTOR_MAP.put(clusterId, jmxMap);
Map<Integer, KafkaVersion> versionMap = KAFKA_VERSION_MAP.getOrDefault(clusterId, new ConcurrentHashMap<>());
@@ -539,9 +542,12 @@ public class PhysicalClusterMetadataManager {
}
public static Set<String> getBrokerTopicNum(Long clusterId, Set<Integer> brokerIdSet) {
Set<String> topicNameSet = new HashSet<>();
Map<String, TopicMetadata> metadataMap = TOPIC_METADATA_MAP.get(clusterId);
if (metadataMap == null) {
return new HashSet<>();
}
Set<String> topicNameSet = new HashSet<>();
for (String topicName: metadataMap.keySet()) {
try {
TopicMetadata tm = metadataMap.get(topicName);

68
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/cache/ThreadPool.java vendored
View File

@@ -1,37 +1,63 @@
package com.xiaojukeji.kafka.manager.service.cache;
import com.xiaojukeji.kafka.manager.common.utils.factory.DefaultThreadFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import java.util.concurrent.*;
import javax.annotation.PostConstruct;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* @author zengqiao
* @date 20/8/24
*/
@Service
public class ThreadPool {
private static final ExecutorService COLLECT_METRICS_THREAD_POOL = new ThreadPoolExecutor(
256,
256,
120L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(),
new DefaultThreadFactory("Collect-Metrics-Thread")
);
private static final ExecutorService API_CALL_THREAD_POOL = new ThreadPoolExecutor(
16,
16,
120L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(),
new DefaultThreadFactory("Api-Call-Thread")
);
@Value(value = "${thread-pool.collect-metrics.thread-num:256}")
private Integer collectMetricsThreadNum;
public static void submitCollectMetricsTask(Runnable collectMetricsTask) {
COLLECT_METRICS_THREAD_POOL.submit(collectMetricsTask);
@Value(value = "${thread-pool.collect-metrics.queue-size:10000}")
private Integer collectMetricsQueueSize;
@Value(value = "${thread-pool.api-call.thread-num:16}")
private Integer apiCallThreadNum;
@Value(value = "${thread-pool.api-call.queue-size:10000}")
private Integer apiCallQueueSize;
private ThreadPoolExecutor collectMetricsThreadPool;
private ThreadPoolExecutor apiCallThreadPool;
@PostConstruct
public void init() {
collectMetricsThreadPool = new ThreadPoolExecutor(
collectMetricsThreadNum,
collectMetricsThreadNum,
120L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(collectMetricsQueueSize),
new DefaultThreadFactory("TaskThreadPool")
);
apiCallThreadPool = new ThreadPoolExecutor(
apiCallThreadNum,
apiCallThreadNum,
120L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(apiCallQueueSize),
new DefaultThreadFactory("ApiThreadPool")
);
}
public static void submitApiCallTask(Runnable apiCallTask) {
API_CALL_THREAD_POOL.submit(apiCallTask);
public void submitCollectMetricsTask(Long clusterId, Runnable collectMetricsTask) {
collectMetricsThreadPool.submit(collectMetricsTask);
}
public void submitApiCallTask(Long clusterId, Runnable apiCallTask) {
apiCallThreadPool.submit(apiCallTask);
}
}

7
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/gateway/impl/GatewayConfigServiceImpl.java
View File

@@ -185,7 +185,8 @@ public class GatewayConfigServiceImpl implements GatewayConfigService {
List<GatewayConfigDO> gatewayConfigDOList = gatewayConfigDao.getByConfigType(gatewayConfigDO.getType());
Long version = 1L;
for (GatewayConfigDO elem: gatewayConfigDOList) {
if (elem.getVersion() > version) {
if (elem.getVersion() >= version) {
// 大于等于的情况下，都需要+1
version = elem.getVersion() + 1L;
}
}
@@ -204,6 +205,7 @@ public class GatewayConfigServiceImpl implements GatewayConfigService {
@Override
public Result deleteById(Long id) {
try {
// TODO 删除的时候，不能直接删，也需要变更一下version
if (gatewayConfigDao.deleteById(id) > 0) {
return Result.buildSuc();
}
@@ -232,7 +234,8 @@ public class GatewayConfigServiceImpl implements GatewayConfigService {
List<GatewayConfigDO> gatewayConfigDOList = gatewayConfigDao.getByConfigType(newGatewayConfigDO.getType());
Long version = 1L;
for (GatewayConfigDO elem: gatewayConfigDOList) {
if (elem.getVersion() > version) {
if (elem.getVersion() >= version) {
// 大于等于的情况下，都需要+1
version = elem.getVersion() + 1L;
}
}

5
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/impl/BrokerServiceImpl.java
View File

@@ -61,6 +61,9 @@ public class BrokerServiceImpl implements BrokerService {
@Autowired
private PhysicalClusterMetadataManager physicalClusterMetadataManager;
@Autowired
private ThreadPool threadPool;
@Override
public ClusterBrokerStatus getClusterBrokerStatus(Long clusterId) {
// 副本同步状态
@@ -201,7 +204,7 @@ public class BrokerServiceImpl implements BrokerService {
return getBrokerMetricsFromJmx(clusterId, brokerId, metricsCode);
}
});
ThreadPool.submitApiCallTask(taskList[i]);
threadPool.submitApiCallTask(clusterId, taskList[i]);
}
List<BrokerMetrics> metricsList = new ArrayList<>(brokerIdSet.size());
for (int i = 0; i < brokerIdList.size(); i++) {

5
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/impl/ExpertServiceImpl.java
View File

@@ -22,6 +22,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.*;
import java.util.regex.Pattern;
/**
* @author zengqiao
@@ -240,9 +241,11 @@ public class ExpertServiceImpl implements ExpertService {
return new ArrayList<>();
}
//获取满足条件的过期Topic
List<TopicExpiredDO> filteredExpiredTopicList = new ArrayList<>();
for (TopicExpiredDO elem: expiredTopicList) {
if (config.getIgnoreClusterIdList().contains(elem.getClusterId())) {
//判定是否为忽略Cluster或者判定是否为忽略Topic名，使用正则来过滤理论上不属于过期的Topic
if (config.getIgnoreClusterIdList().contains(elem.getClusterId()) || Pattern.matches(config.getFilterRegex(), elem.getTopicName())) {
continue;
}
filteredExpiredTopicList.add(elem);

7
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/impl/JmxServiceImpl.java
View File

@@ -39,6 +39,9 @@ public class JmxServiceImpl implements JmxService {
@Autowired
private PhysicalClusterMetadataManager physicalClusterMetadataManager;
@Autowired
private ThreadPool threadPool;
@Override
public BrokerMetrics getBrokerMetrics(Long clusterId, Integer brokerId, Integer metricsCode) {
if (clusterId == null || brokerId == null || metricsCode == null) {
@@ -98,7 +101,7 @@ public class JmxServiceImpl implements JmxService {
);
}
});
ThreadPool.submitCollectMetricsTask(taskList[i]);
threadPool.submitCollectMetricsTask(clusterId, taskList[i]);
}
List<TopicMetrics> metricsList = new ArrayList<>();
@@ -305,7 +308,7 @@ public class JmxServiceImpl implements JmxService {
return metricsList;
}
});
ThreadPool.submitCollectMetricsTask(taskList[i]);
threadPool.submitCollectMetricsTask(clusterId, taskList[i]);
}
Map<String, TopicMetrics> metricsMap = new HashMap<>();

4
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/impl/RegionServiceImpl.java
View File

@@ -2,6 +2,8 @@ package com.xiaojukeji.kafka.manager.service.service.impl;
import com.xiaojukeji.kafka.manager.common.entity.ResultStatus;
import com.xiaojukeji.kafka.manager.common.entity.pojo.RegionDO;
import com.xiaojukeji.kafka.manager.common.events.RegionCreatedEvent;
import com.xiaojukeji.kafka.manager.common.utils.SpringTool;
import com.xiaojukeji.kafka.manager.common.utils.ValidateUtils;
import com.xiaojukeji.kafka.manager.common.zookeeper.znode.brokers.TopicMetadata;
import com.xiaojukeji.kafka.manager.dao.RegionDao;
@@ -59,6 +61,8 @@ public class RegionServiceImpl implements RegionService {
return ResultStatus.BROKER_NOT_EXIST;
}
if (regionDao.insert(regionDO) > 0) {
// 发布region创建事件
SpringTool.publish(new RegionCreatedEvent(this, regionDO));
return ResultStatus.SUCCESS;
}
} catch (DuplicateKeyException e) {

7
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/service/impl/TopicServiceImpl.java
View File

@@ -87,6 +87,9 @@ public class TopicServiceImpl implements TopicService {
@Autowired
private AbstractHealthScoreStrategy healthScoreStrategy;
@Autowired
private KafkaClientPool kafkaClientPool;
@Override
public List<TopicMetricsDO> getTopicMetricsFromDB(Long clusterId, String topicName, Date startTime, Date endTime) {
try {
@@ -340,7 +343,7 @@ public class TopicServiceImpl implements TopicService {
Map<TopicPartition, Long> topicPartitionLongMap = new HashMap<>();
KafkaConsumer kafkaConsumer = null;
try {
kafkaConsumer = KafkaClientPool.borrowKafkaConsumerClient(clusterDO);
kafkaConsumer = kafkaClientPool.borrowKafkaConsumerClient(clusterDO);
if ((offsetPosEnum.getCode() & OffsetPosEnum.END.getCode()) > 0) {
topicPartitionLongMap = kafkaConsumer.endOffsets(topicPartitionList);
} else if ((offsetPosEnum.getCode() & OffsetPosEnum.BEGINNING.getCode()) > 0) {
@@ -538,7 +541,7 @@ public class TopicServiceImpl implements TopicService {
List<PartitionOffsetDTO> partitionOffsetDTOList = new ArrayList<>();
try {
kafkaConsumer = KafkaClientPool.borrowKafkaConsumerClient(clusterDO);
kafkaConsumer = kafkaClientPool.borrowKafkaConsumerClient(clusterDO);
Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestampMap = kafkaConsumer.offsetsForTimes(timestampsToSearch);
if (offsetAndTimestampMap == null) {
return new ArrayList<>();

5
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/strategy/healthscore/DidiHealthScoreStrategy.java
View File

@@ -45,6 +45,9 @@ public class DidiHealthScoreStrategy extends AbstractHealthScoreStrategy {
@Autowired
private JmxService jmxService;
@Autowired
private ThreadPool threadPool;
@Override
public Integer calBrokerHealthScore(Long clusterId, Integer brokerId) {
BrokerMetadata brokerMetadata = PhysicalClusterMetadataManager.getBrokerMetadata(clusterId, brokerId);
@@ -125,7 +128,7 @@ public class DidiHealthScoreStrategy extends AbstractHealthScoreStrategy {
return calBrokerHealthScore(clusterId, brokerId);
}
});
ThreadPool.submitApiCallTask(taskList[i]);
threadPool.submitApiCallTask(clusterId, taskList[i]);
}
Integer topicHealthScore = HEALTH_SCORE_HEALTHY;

37
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/utils/ConfigUtils.java
View File

@@ -1,5 +1,6 @@
package com.xiaojukeji.kafka.manager.service.utils;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
@@ -8,38 +9,18 @@ import org.springframework.stereotype.Service;
* @author zengqiao
* @date 20/4/26
*/
@Data
@Service("configUtils")
public class ConfigUtils {
@Value(value = "${custom.idc}")
private ConfigUtils() {
}
@Value(value = "${custom.idc:cn}")
private String idc;
@Value(value = "${spring.profiles.active}")
@Value(value = "${spring.profiles.active:dev}")
private String kafkaManagerEnv;
@Value(value = "${custom.store-metrics-task.save-days}")
private Long maxMetricsSaveDays;
public String getIdc() {
return idc;
}
public void setIdc(String idc) {
this.idc = idc;
}
public String getKafkaManagerEnv() {
return kafkaManagerEnv;
}
public void setKafkaManagerEnv(String kafkaManagerEnv) {
this.kafkaManagerEnv = kafkaManagerEnv;
}
public Long getMaxMetricsSaveDays() {
return maxMetricsSaveDays;
}
public void setMaxMetricsSaveDays(Long maxMetricsSaveDays) {
this.maxMetricsSaveDays = maxMetricsSaveDays;
}
@Value(value = "${spring.application.version:unknown}")
private String applicationVersion;
}

11
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/zookeeper/BrokerStateListener.java
View File

@@ -74,15 +74,10 @@ public class BrokerStateListener implements StateChangeListener {
BrokerMetadata brokerMetadata = null;
try {
brokerMetadata = zkConfig.get(ZkPathUtil.getBrokerIdNodePath(brokerId), BrokerMetadata.class);
if (!brokerMetadata.getEndpoints().isEmpty()) {
String endpoint = brokerMetadata.getEndpoints().get(0);
int idx = endpoint.indexOf("://");
endpoint = endpoint.substring(idx + "://".length());
idx = endpoint.indexOf(":");
brokerMetadata.setHost(endpoint.substring(0, idx));
brokerMetadata.setPort(Integer.parseInt(endpoint.substring(idx + 1)));
}
// 解析并更新本次存储的broker元信息
BrokerMetadata.parseAndUpdateBrokerMetadata(brokerMetadata);
brokerMetadata.setClusterId(clusterId);
brokerMetadata.setBrokerId(brokerId);
PhysicalClusterMetadataManager.putBrokerMetadata(clusterId, brokerId, brokerMetadata, jmxConfig);

28
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/zookeeper/ControllerStateListener.java
View File

@@ -19,13 +19,13 @@ import org.springframework.dao.DuplicateKeyException;
* @date 20/5/14
*/
public class ControllerStateListener implements StateChangeListener {
private final static Logger LOGGER = LoggerFactory.getLogger(ControllerStateListener.class);
private static final Logger LOGGER = LoggerFactory.getLogger(ControllerStateListener.class);
private Long clusterId;
private final Long clusterId;
private ZkConfigImpl zkConfig;
private final ZkConfigImpl zkConfig;
private ControllerDao controllerDao;
private final ControllerDao controllerDao;
public ControllerStateListener(Long clusterId, ZkConfigImpl zkConfig, ControllerDao controllerDao) {
this.clusterId = clusterId;
@@ -35,8 +35,11 @@ public class ControllerStateListener implements StateChangeListener {
@Override
public void init() {
if (!checkNodeExist()) {
LOGGER.warn("kafka-controller data not exist, clusterId:{}.", clusterId);
return;
}
processControllerChange();
return;
}
@Override
@@ -49,12 +52,21 @@ public class ControllerStateListener implements StateChangeListener {
break;
}
} catch (Exception e) {
LOGGER.error("process controller state change failed, clusterId:{} state:{} path:{}.",
clusterId, state, path, e);
LOGGER.error("process controller state change failed, clusterId:{} state:{} path:{}.", clusterId, state, path, e);
}
}
private void processControllerChange(){
private boolean checkNodeExist() {
try {
return zkConfig.checkPathExists(ZkPathUtil.CONTROLLER_ROOT_NODE);
} catch (Exception e) {
LOGGER.error("init kafka-controller data failed, clusterId:{}.", clusterId, e);
}
return false;
}
private void processControllerChange() {
LOGGER.warn("init controllerData or controller change, clusterId:{}.", clusterId);
ControllerData controllerData = null;
try {

8
kafka-manager-core/src/main/java/com/xiaojukeji/kafka/manager/service/zookeeper/TopicStateListener.java
View File

@@ -10,6 +10,7 @@ import com.xiaojukeji.kafka.manager.service.cache.ThreadPool;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import java.util.HashSet;
import java.util.List;
@@ -28,9 +29,12 @@ public class TopicStateListener implements StateChangeListener {
private ZkConfigImpl zkConfig;
public TopicStateListener(Long clusterId, ZkConfigImpl zkConfig) {
private ThreadPool threadPool;
public TopicStateListener(Long clusterId, ZkConfigImpl zkConfig, ThreadPool threadPool) {
this.clusterId = clusterId;
this.zkConfig = zkConfig;
this.threadPool = threadPool;
}
@Override
@@ -47,7 +51,7 @@ public class TopicStateListener implements StateChangeListener {
return null;
}
});
ThreadPool.submitCollectMetricsTask(taskList[i]);
threadPool.submitCollectMetricsTask(clusterId, taskList[i]);
}
} catch (Exception e) {
LOGGER.error("init topics metadata failed, clusterId:{}.", clusterId, e);