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

1、增加AdminClient的ClientID； 2、关闭时，增加超时时间； 3、增加关闭错误的日志；
didi · Aug 2, 2023 · db40a5c · db40a5c
1 parent 55161e4
commit db40a5c
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 4 deletions.
diff --git a/.../main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupServiceImpl.java b/.../main/java/com/xiaojukeji/know/streaming/km/core/service/group/impl/GroupServiceImpl.java
@@ -78,6 +78,7 @@ public List<String> listGroupsFromKafka(ClusterPhy clusterPhy) throws AdminOpera
             }
 
             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 KSGroupDescription getGroupDescriptionFromKafka(ClusterPhy clusterPhy, St
             }
 
             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);
 

diff --git a/...ce/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/KafkaAdminClient.java b/...ce/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/KafkaAdminClient.java
@@ -12,6 +12,7 @@
 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 @@ private void closeKafkaAdminClient(Long clusterPhyId) {
 
             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 @@ private AdminClient createKafkaAdminClient(Long clusterPhyId, String bootstrapSe
 
             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,25 +129,27 @@ private AdminClient createKafkaAdminClient(Long clusterPhyId, String bootstrapSe
         } catch (Exception e) {
             LOGGER.error("create kafka AdminClient failed, clusterPhyId:{} props:{}", clusterPhyId, props, e);
 
-            this.closeAdminClientList(adminClientList);
+            this.closeAdminClientList(clusterPhyId, adminClientList);
         } finally {
             modifyClientMapLock.unlock();
         }
 
         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;
         }
 
         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;
             }
         }