From 0a6e9b7633abe611e150935fc64650f91e9d79ee Mon Sep 17 00:00:00 2001 From: EricZeng Date: Thu, 6 Jul 2023 15:39:51 +0800 Subject: [PATCH] =?UTF-8?q?[Optimize]Jmx=E7=9B=B8=E5=85=B3=E6=97=A5?= =?UTF-8?q?=E5=BF=97=E4=BC=98=E5=8C=96=20(#1082)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1、统一Jmx客户端相关日志格式； 2、增加创建的Jmx-Connector时，所使用的信息； 3、优化日志级别； --- .../streaming/km/common/jmx/JmxConnectorWrap.java | 12 +++++++++++- .../km/persistence/connect/ConnectJMXClient.java | 11 +++++++---- .../km/persistence/kafka/KafkaJMXClient.java | 4 ++-- 3 files changed, 20 insertions(+), 7 deletions(-) diff --git a/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/jmx/JmxConnectorWrap.java b/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/jmx/JmxConnectorWrap.java index 071072ad..366fe947 100644 --- a/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/jmx/JmxConnectorWrap.java +++ b/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/jmx/JmxConnectorWrap.java @@ -46,7 +46,12 @@ public class JmxConnectorWrap { private JmxAuthConfig jmxConfig; public JmxConnectorWrap(String clientLogIdent, Long brokerStartupTime, String jmxHost, Integer jmxPort, JmxAuthConfig jmxConfig) { - this.clientLogIdent=clientLogIdent; + LOGGER.info( + "method=JmxConnectorWrap||clientLogIdent={}||brokerStartupTime={}||jmxHost={}||jmxPort={}||jmxConfig={}||msg=start construct JmxWrap.", + clientLogIdent, brokerStartupTime, jmxHost, jmxPort, jmxConfig + ); + + this.clientLogIdent = clientLogIdent; this.brokerStartupTime = brokerStartupTime; this.jmxHost = jmxHost; this.jmxPort = (jmxPort == null? JmxEnum.UNKNOWN.getPort() : jmxPort); @@ -160,6 +165,11 @@ public class JmxConnectorWrap { if (jmxConnector != null) { return true; } + LOGGER.info( + "method=createJmxConnector||clientLogIdent={}||brokerStartupTime={}||jmxHost={}||jmxPort={}||jmxConfig={}||msg=start create jmx connector.", + clientLogIdent, brokerStartupTime, jmxHost, jmxPort, jmxConfig + ); + String jmxUrl = String.format("service:jmx:rmi:///jndi/rmi://%s:%d/jmxrmi", jmxHost, jmxPort); try { Map environment = new HashMap(); diff --git a/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/connect/ConnectJMXClient.java b/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/connect/ConnectJMXClient.java index 300243b7..ea7e38c5 100644 --- a/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/connect/ConnectJMXClient.java +++ b/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/connect/ConnectJMXClient.java @@ -78,16 +78,15 @@ public class ConnectJMXClient extends AbstractConnectClusterChangeHandler { return jmxConnectorWrap; } - log.debug("method=createJmxConnectorWrap||connectClusterId={}||workerId={}||msg=create JmxConnectorWrap starting", connectCluster.getId(), workerId); + log.info("method=createJmxConnectorWrap||connectClusterId={}||workerId={}||msg=create JmxConnectorWrap starting", connectCluster.getId(), workerId); JmxConfig jmxConfig = ConvertUtil.str2ObjByJson(connectCluster.getJmxProperties(), JmxConfig.class); if (jmxConfig == null) { jmxConfig = new JmxConfig(); } - jmxConnectorWrap = new JmxConnectorWrap( - "connectClusterId: " + connectCluster.getId() + " workerId: " + workerId, + String.format("clusterPhyId=%s,workerId=%s", connectCluster.getId(), workerId), null, connectWorker.getHost(), jmxConfig.getFinallyJmxPort(workerId, connectWorker.getJmxPort()), @@ -97,12 +96,16 @@ public class ConnectJMXClient extends AbstractConnectClusterChangeHandler { Map workerMap = JMX_MAP.getOrDefault(connectCluster.getId(), new ConcurrentHashMap<>()); workerMap.put(workerId, jmxConnectorWrap); JMX_MAP.put(connectCluster.getId(), workerMap); + + log.info("method=createJmxConnectorWrap||clusterPhyId={}||workerId={}||msg=create JmxConnectorWrap success", connectCluster.getId(), workerId); + return jmxConnectorWrap; } catch (Exception e) { - log.debug("method=createJmxConnectorWrap||connectClusterId={}||workerId={}||msg=create JmxConnectorWrap failed||errMsg=exception||", connectCluster.getId(), workerId, e); + log.error("method=createJmxConnectorWrap||connectClusterId={}||workerId={}||msg=create JmxConnectorWrap failed||errMsg=exception||", connectCluster.getId(), workerId, e); } finally { modifyClientMapLock.unlock(); } + return null; } diff --git a/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/KafkaJMXClient.java b/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/KafkaJMXClient.java index e759da60..8c8ca9cd 100644 --- a/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/KafkaJMXClient.java +++ b/km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/kafka/KafkaJMXClient.java @@ -151,7 +151,7 @@ public class KafkaJMXClient extends AbstractClusterLoadedChangedHandler { return jmxMap; } - log.debug("method=createJmxConnectorWrap||clusterPhyId={}||brokerId={}||msg=create JmxConnectorWrap starting", clusterPhy.getId(), brokerId); + log.info("method=createJmxConnectorWrap||clusterPhyId={}||brokerId={}||msg=create JmxConnectorWrap starting", clusterPhy.getId(), brokerId); JmxConfig jmxConfig = ConvertUtil.str2ObjByJson(clusterPhy.getJmxProperties(), JmxConfig.class); if (jmxConfig == null) { @@ -159,7 +159,7 @@ public class KafkaJMXClient extends AbstractClusterLoadedChangedHandler { } JmxConnectorWrap jmxConnectorWrap = new JmxConnectorWrap( - "clusterPhyId: " + clusterPhy.getId() + " brokerId: " + brokerId, + String.format("clusterPhyId=%s,brokerId=%d", clusterPhy.getId(), brokerId), broker.getStartTimestamp(), broker.getJmxHost(jmxConfig.getUseWhichEndpoint()), jmxConfig.getFinallyJmxPort(String.valueOf(brokerId), broker.getJmxPort()),