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

[Optimize]增加AdminClient观测信息 (#1111)

Browse Source 
1、增加AdminClient的ClientID；
2、关闭时，增加超时时间；
3、增加关闭错误的日志；
This commit is contained in:
EricZeng
2023-08-02 21:19:03 +08:00
committed by GitHub
parent 55161e439a
commit db40a5cd0a
2 changed files with 12 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupServiceImpl.java
View File

@@ -78,6 +78,7 @@ public class GroupServiceImpl extends BaseKafkaVersionControlService implements
}
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, clusterPhy.getBootstrapServers());
props.put(AdminClientConfig.CLIENT_ID_CONFIG, String.format("KSPartialAdminClient||clusterPhyId=%d", clusterPhy.getId()));
adminClient = KSPartialKafkaAdminClient.create(props);
KSListGroupsResult listConsumerGroupsResult = adminClient.listConsumerGroups(
@@ -178,6 +179,7 @@ public class GroupServiceImpl extends BaseKafkaVersionControlService implements
}
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, clusterPhy.getBootstrapServers());
props.put(AdminClientConfig.CLIENT_ID_CONFIG, String.format("KSPartialAdminClient||clusterPhyId=%d", clusterPhy.getId()));
adminClient = KSPartialKafkaAdminClient.create(props);

14
km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/KafkaAdminClient.java
View File

@@ -12,6 +12,7 @@ import org.apache.kafka.clients.admin.AdminClientConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -76,10 +77,12 @@ public class KafkaAdminClient extends AbstractClusterLoadedChangedHandler {
LOGGER.info("close kafka AdminClient starting, clusterPhyId:{}", clusterPhyId);
boolean allSuccess = this.closeAdminClientList(adminClientList);
boolean allSuccess = this.closeAdminClientList(clusterPhyId, adminClientList);
if (allSuccess) {
LOGGER.info("close kafka AdminClient success, clusterPhyId:{}", clusterPhyId);
} else {
LOGGER.error("close kafka AdminClient exist failed and can ignore this error, clusterPhyId:{}", clusterPhyId);
}
} catch (Exception e) {
LOGGER.error("close kafka AdminClient failed, clusterPhyId:{}", clusterPhyId, e);
@@ -116,6 +119,7 @@ public class KafkaAdminClient extends AbstractClusterLoadedChangedHandler {
adminClientList = new ArrayList<>();
for (int i = 0; i < clientCnt; ++i) {
props.put(AdminClientConfig.CLIENT_ID_CONFIG, String.format("ApacheAdminClient||clusterPhyId=%d||Cnt=%d", clusterPhyId, i));
adminClientList.add(AdminClient.create(props));
}
@@ -125,7 +129,7 @@ public class KafkaAdminClient extends AbstractClusterLoadedChangedHandler {
} catch (Exception e) {
LOGGER.error("create kafka AdminClient failed, clusterPhyId:{} props:{}", clusterPhyId, props, e);
this.closeAdminClientList(adminClientList);
this.closeAdminClientList(clusterPhyId, adminClientList);
} finally {
modifyClientMapLock.unlock();
}
@@ -133,7 +137,7 @@ public class KafkaAdminClient extends AbstractClusterLoadedChangedHandler {
return KAFKA_ADMIN_CLIENT_MAP.get(clusterPhyId).get((int)(System.currentTimeMillis() % clientCnt));
}
private boolean closeAdminClientList(List<AdminClient> adminClientList) {
private boolean closeAdminClientList(Long clusterPhyId, List<AdminClient> adminClientList) {
if (adminClientList == null) {
return true;
}
@@ -141,9 +145,11 @@ public class KafkaAdminClient extends AbstractClusterLoadedChangedHandler {
boolean allSuccess = true;
for (AdminClient adminClient: adminClientList) {
try {
adminClient.close();
// 关闭客户端，超时时间为30秒
adminClient.close(Duration.ofSeconds(30));
} catch (Exception e) {
// ignore
LOGGER.error("close kafka AdminClient exist failed, clusterPhyId:{}", clusterPhyId, e);
allSuccess = false;
}
}