8x48/KnowStreaming
1
0
Fork 0
mirror of https://github.com/didi/KnowStreaming.git synced 2026-01-04 20:02:07 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

JmxConnectorWrap's log add cluster and broker info

Browse Source
This commit is contained in:
zengqiao
2022-01-11 11:45:28 +08:00
parent 44ea896de8
commit 80785ce072
2 changed files with 16 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
kafka-manager-common/src/main/java/com/xiaojukeji/kafka/manager/common/utils/jmx/JmxConnectorWrap.java
View File

@@ -14,7 +14,6 @@ import javax.naming.Context;
import javax.rmi.ssl.SslRMIClientSocketFactory; import javax.rmi.ssl.SslRMIClientSocketFactory;
import java.io.IOException; import java.io.IOException;
import java.net.MalformedURLException; import java.net.MalformedURLException;
import java.util.Arrays;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map; import java.util.Map;
import java.util.Set; import java.util.Set;
@@ -26,19 +25,25 @@ import java.util.concurrent.atomic.AtomicInteger;
* @date 2015/11/9. * @date 2015/11/9.
*/ */
public class JmxConnectorWrap { public class JmxConnectorWrap {
private final static Logger LOGGER = LoggerFactory.getLogger(JmxConnectorWrap.class); private static final Logger LOGGER = LoggerFactory.getLogger(JmxConnectorWrap.class);
private String host; private final Long physicalClusterId;
private int port; private final Integer brokerId;
private final String host;
private final int port;
private JMXConnector jmxConnector; private JMXConnector jmxConnector;
private AtomicInteger atomicInteger; private final AtomicInteger atomicInteger;
private JmxConfig jmxConfig; private JmxConfig jmxConfig;
public JmxConnectorWrap(String host, int port, JmxConfig jmxConfig) { public JmxConnectorWrap(Long physicalClusterId, Integer brokerId, String host, int port, JmxConfig jmxConfig) {
this.physicalClusterId = physicalClusterId;
this.brokerId = brokerId;
this.host = host; this.host = host;
this.port = port; this.port = port;
this.jmxConfig = jmxConfig; this.jmxConfig = jmxConfig;
@@ -68,7 +73,7 @@ public class JmxConnectorWrap {
try { try {
jmxConnector.close(); jmxConnector.close();
} catch (IOException e) { } catch (IOException e) {
LOGGER.warn("close JmxConnector exception, host:{} port:{}.", host, port, e); LOGGER.warn("close JmxConnector exception, physicalClusterId:{} brokerId:{} host:{} port:{}.", physicalClusterId, brokerId, host, port, e);
} }
} }
@@ -91,12 +96,12 @@ public class JmxConnectorWrap {
} }
jmxConnector = JMXConnectorFactory.connect(new JMXServiceURL(jmxUrl), environment); jmxConnector = JMXConnectorFactory.connect(new JMXServiceURL(jmxUrl), environment);
LOGGER.info("JMX connect success, host:{} port:{}.", host, port); LOGGER.info("JMX connect success, physicalClusterId:{} brokerId:{} host:{} port:{}.", physicalClusterId, brokerId, host, port);
return true; return true;
} catch (MalformedURLException e) { } catch (MalformedURLException e) {
LOGGER.error("JMX url exception, host:{} port:{} jmxUrl:{}", host, port, jmxUrl, e); LOGGER.error("JMX url exception, physicalClusterId:{} brokerId:{} host:{} port:{} jmxUrl:{}", physicalClusterId, brokerId, host, port, jmxUrl, e);
} catch (Exception e) { } catch (Exception e) {
LOGGER.error("JMX connect exception, host:{} port:{}.", host, port, e); LOGGER.error("JMX connect exception, physicalClusterId:{} brokerId:{} host:{} port:{}.", physicalClusterId, brokerId, host, port, e);
} }
return false; return false;
} }

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

@@ -314,7 +314,7 @@ public class PhysicalClusterMetadataManager {
metadataMap.put(brokerId, brokerMetadata); metadataMap.put(brokerId, brokerMetadata);
Map<Integer, JmxConnectorWrap> jmxMap = JMX_CONNECTOR_MAP.getOrDefault(clusterId, new ConcurrentHashMap<>()); 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); JMX_CONNECTOR_MAP.put(clusterId, jmxMap);
Map<Integer, KafkaVersion> versionMap = KAFKA_VERSION_MAP.getOrDefault(clusterId, new ConcurrentHashMap<>()); Map<Integer, KafkaVersion> versionMap = KAFKA_VERSION_MAP.getOrDefault(clusterId, new ConcurrentHashMap<>());