From beb754aaaaef3950ee36c033f6024e7315e850c1 Mon Sep 17 00:00:00 2001
From: zengqiao <zengqiao@didiglobal.com>
Date: Tue, 20 Sep 2022 14:22:06 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8DJMX=E8=BF=9E=E6=8E=A5?=
 =?UTF-8?q?=E8=A2=AB=E5=85=B3=E9=97=AD=EF=BC=8C=E6=8A=9B=E5=87=BAIOExcepti?=
 =?UTF-8?q?on=E5=90=8E=EF=BC=8C=E6=9C=AA=E8=BF=9B=E8=A1=8C=E8=BF=9E?=
 =?UTF-8?q?=E6=8E=A5=E9=87=8D=E5=BB=BA=E7=9A=84=E9=97=AE=E9=A2=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../km/common/jmx/JmxConnectorWrap.java       | 39 +++++++++++++++++++
 1 file changed, 39 insertions(+)

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 0fb65589..ca7c01c4 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
@@ -90,6 +90,8 @@ public class JmxConnectorWrap {
         }
         try {
             jmxConnector.close();
+
+            jmxConnector = null;
         } catch (IOException e) {
             LOGGER.warn("close JmxConnector exception, physicalClusterId:{} brokerId:{} host:{} port:{}.", physicalClusterId, brokerId, host, port, e);
         }
@@ -105,6 +107,11 @@ public class JmxConnectorWrap {
             acquire();
             MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();
             return mBeanServerConnection.getAttribute(name, attribute);
+        } catch (IOException ioe) {
+            // 如果是因为连接断开，则进行重新连接，并抛出异常
+            reInitDueIOException();
+
+            throw ioe;
         } finally {
             atomicInteger.incrementAndGet();
         }
@@ -120,6 +127,11 @@ public class JmxConnectorWrap {
             acquire();
             MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();
             return mBeanServerConnection.getAttributes(name, attributes);
+        } catch (IOException ioe) {
+            // 如果是因为连接断开，则进行重新连接，并抛出异常
+            reInitDueIOException();
+
+            throw ioe;
         } finally {
             atomicInteger.incrementAndGet();
         }
@@ -131,6 +143,11 @@ public class JmxConnectorWrap {
             acquire();
             MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();
             return mBeanServerConnection.queryNames(name, query);
+        } catch (IOException ioe) {
+            // 如果是因为连接断开，则进行重新连接，并抛出异常
+            reInitDueIOException();
+
+            throw ioe;
         } finally {
             atomicInteger.incrementAndGet();
         }
@@ -186,4 +203,26 @@ public class JmxConnectorWrap {
             }
         }
     }
+
+    private synchronized void reInitDueIOException() {
+        try {
+            if (jmxConnector == null) {
+                return;
+            }
+
+            // 检查是否正常
+            jmxConnector.getConnectionId();
+
+            // 如果正常则直接返回
+            return;
+        } catch (Exception e) {
+            // ignore
+        }
+
+        // 关闭旧的
+        this.close();
+
+        // 重新创建
+        this.checkJmxConnectionAndInitIfNeed();
+    }
 }