Repository: skiptests/astraea
Pull request #460 (2022-07-06 18:46:47)

Introduce local MBeanClient
我們即將新增一種新的metrics評量方式 - 監控producer local metrics來評估合適的節點，為了達到這個目的，我們需要將local JVM metrics串接到`MBeanClient`，這隻PR達成這第一步，接下來會開始引入可以用local metrics來評估的各種cost functions。這個方法的最大好處是使用者“不需要"輸入JMX port等資訊一樣可以享受cost functions帶來的好處


Patch:
diff --git a/app/src/main/java/org/astraea/app/metrics/jmx/MBeanClient.java b/app/src/main/java/org/astraea/app/metrics/jmx/MBeanClient.java
index c46462df..23ad00e1 100644
--- a/app/src/main/java/org/astraea/app/metrics/jmx/MBeanClient.java
+++ b/app/src/main/java/org/astraea/app/metrics/jmx/MBeanClient.java
@@ -54,7 +54,11 @@ public interface MBeanClient extends AutoCloseable {
   }
 
   static MBeanClient of(JMXServiceURL url) {
-    return new MBeanClientImpl(url);
+    return MBeanClientImpl.remote(url);
+  }
+
+  static MBeanClient local() {
+    return MBeanClientImpl.local();
   }
 
   /**
diff --git a/app/src/main/java/org/astraea/app/metrics/jmx/MBeanClientImpl.java b/app/src/main/java/org/astraea/app/metrics/jmx/MBeanClientImpl.java
index 67c44c4a..36260b34 100644
--- a/app/src/main/java/org/astraea/app/metrics/jmx/MBeanClientImpl.java
+++ b/app/src/main/java/org/astraea/app/metrics/jmx/MBeanClientImpl.java
@@ -18,55 +18,80 @@ package org.astraea.app.metrics.jmx;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.lang.management.ManagementFactory;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.NoSuchElementException;
-import java.util.Objects;
-import java.util.Set;
 import java.util.stream.Collectors;
-import javax.management.Attribute;
 import javax.management.InstanceNotFoundException;
 import javax.management.IntrospectionException;
 import javax.management.MBeanFeatureInfo;
-import javax.management.MBeanInfo;
 import javax.management.MBeanServerConnection;
 import javax.management.ObjectInstance;
 import javax.management.ReflectionException;
-import javax.management.remote.JMXConnector;
 import javax.management.remote.JMXConnectorFactory;
 import javax.management.remote.JMXServiceURL;
 import org.astraea.app.common.Utils;
 
-class MBeanClientImpl implements MBeanClient {
+abstract class MBeanClientImpl implements MBeanClient {
 
-  private final JMXServiceURL jmxServiceURL;
-  private final JMXConnector jmxConnector;
-  private final MBeanServerConnection mBeanServerConnection;
-  private boolean isClosed;
-
-  MBeanClientImpl(JMXServiceURL jmxServiceURL) {
+  static MBeanClientImpl remote(JMXServiceURL jmxServiceURL) {
     try {
-      this.jmxServiceURL = Objects.requireNonNull(jmxServiceURL);
-      this.jmxConnector = JMXConnectorFactory.connect(jmxServiceURL);
-      this.isClosed = false;
-      this.mBeanServerConnection = jmxConnector.getMBeanServerConnection();
+      var jmxConnector = JMXConnectorFactory.connect(jmxServiceURL);
+      return new MBeanClientImpl(jmxConnector.getMBeanServerConnection()) {
+        @Override
+        public String host() {
+          return jmxServiceURL.getHost();
+        }
+
+        @Override
+        public int port() {
+          return jmxServiceURL.getPort();
+        }
+
+        @Override
+        public void close() {
+          Utils.packException(jmxConnector::close);
+        }
+      };
     } catch (IOException e) {
       throw new UncheckedIOException(e);
     }
   }
 
+  static MBeanClientImpl local() {
+    return new MBeanClientImpl(ManagementFactory.getPlatformMBeanServer()) {
+      @Override
+      public String host() {
+        return Utils.hostname();
+      }
+
+      @Override
+      public int port() {
+        return -1;
+      }
+
+      @Override
+      public void close() {}
+    };
+  }
+
+  private final MBeanServerConnection connection;
+
+  MBeanClientImpl(MBeanServerConnection connection) {
+    this.connection = connection;
+  }
+
   @Override
   public BeanObject queryBean(BeanQuery beanQuery) {
-    ensureConnected();
     try {
       // ask for MBeanInfo
-      MBeanInfo mBeanInfo = mBeanServerConnection.getMBeanInfo(beanQuery.objectName());
+      var mBeanInfo = connection.getMBeanInfo(beanQuery.objectName());
 
       // create a list builder all available attributes name
-      List<String> attributeName =
+      var attributeName =
           Arrays.stream(mBeanInfo.getAttributes())
               .map(MBeanFeatureInfo::getName)
               .collect(Collectors.toList());
@@ -84,32 +109,30 @@ class MBeanClientImpl implements MBeanClient {
 
   @Override
   public BeanObject queryBean(BeanQuery beanQuery, Collection<String> attributeNameCollection) {
-    ensureConnected();
     try {
 
       // fetch attribute value from mbean server
-      String[] attributeNameArray = attributeNameCollection.toArray(new String[0]);
-      List<Attribute> attributeList =
-          mBeanServerConnection.getAttributes(beanQuery.objectName(), attributeNameArray).asList();
+      var attributeNameArray = attributeNameCollection.toArray(new String[0]);
+      var attributeList =
+          connection.getAttributes(beanQuery.objectName(), attributeNameArray).asList();
 
       // collect attribute name & value into a map
-      Map<String, Object> attributes = new HashMap<>();
-      for (Attribute attribute : attributeList) {
-        attributes.put(attribute.getName(), attribute.getValue());
-      }
+      var attributes = new HashMap<String, Object>();
+      attributeList.forEach(attribute -> attributes.put(attribute.getName(), attribute.getValue()));
 
       // according to the javadoc of MBeanServerConnection#getAttributes, the API will
       // ignore any
       // error occurring during the fetch process (for example, attribute not exists). Below code
       // check for such condition and try to figure out what exactly the error is. put it into
       // attributes return result.
-      Set<String> notResolvedAttributes =
+      var notResolvedAttributes =
           Arrays.stream(attributeNameArray)
               .filter(str -> !attributes.containsKey(str))
               .collect(Collectors.toSet());
-      for (String attributeName : notResolvedAttributes) {
-        attributes.put(attributeName, fetchAttributeObjectOrException(beanQuery, attributeName));
-      }
+      notResolvedAttributes.forEach(
+          attributeName ->
+              attributes.put(
+                  attributeName, fetchAttributeObjectOrException(beanQuery, attributeName)));
 
       // collect result, and build a new BeanObject as return result
       return new BeanObject(beanQuery.domainName(), beanQuery.properties(), attributes);
@@ -135,7 +158,7 @@ class MBeanClientImpl implements MBeanClient {
     // exception
     // into their result.
     try {
-      return mBeanServerConnection.getAttribute(beanQuery.objectName(), attributeName);
+      return connection.getAttribute(beanQuery.objectName(), attributeName);
     } catch (IOException e) {
       throw new UncheckedIOException(e);
     } catch (Exception e) {
@@ -145,9 +168,8 @@ class MBeanClientImpl implements MBeanClient {
 
   @Override
   public Collection<BeanObject> queryBeans(BeanQuery beanQuery) {
-    ensureConnected();
     try {
-      return mBeanServerConnection.queryMBeans(beanQuery.objectName(), null).stream()
+      return connection.queryMBeans(beanQuery.objectName(), null).stream()
           .map(ObjectInstance::getObjectName)
           .map(BeanQuery::fromObjectName)
           .map(this::queryBean)
@@ -161,40 +183,9 @@ class MBeanClientImpl implements MBeanClient {
   @Override
   public List<String> listDomains() {
     try {
-      return Arrays.asList(mBeanServerConnection.getDomains());
+      return Arrays.asList(connection.getDomains());
     } catch (IOException e) {
       throw new UncheckedIOException(e);
     }
   }
-
-  /**
-   * The JMX URL of current MBeanClient instance
-   *
-   * @return the origin JMX URL used to initiate MBeanClient
-   */
-  public JMXServiceURL getAddress() {
-    return jmxServiceURL;
-  }
-
-  @Override
-  public String host() {
-    return jmxServiceURL.getHost();
-  }
-
-  @Override
-  public int port() {
-    return jmxServiceURL.getPort();
-  }
-
-  private void ensureConnected() {
-    if (isClosed) throw new IllegalStateException("MBean client is closed");
-  }
-
-  @Override
-  public void close() {
-    if (!isClosed) {
-      isClosed = true;
-      Utils.packException(jmxConnector::close);
-    }
-  }
 }


Test patch:
diff --git a/app/src/test/java/org/astraea/app/metrics/jmx/MBeanClientTest.java b/app/src/test/java/org/astraea/app/metrics/jmx/MBeanClientTest.java
index 1a636e73..b9ffa744 100644
--- a/app/src/test/java/org/astraea/app/metrics/jmx/MBeanClientTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/jmx/MBeanClientTest.java
@@ -45,6 +45,7 @@ import javax.management.remote.JMXConnectorServer;
 import javax.management.remote.JMXConnectorServerFactory;
 import javax.management.remote.JMXServiceURL;
 import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -233,23 +234,6 @@ class MBeanClientTest {
     }
   }
 
-  @Test
-  void testUseClosedClientWillThrowError() {
-    // arrange
-    var client = MBeanClient.of(jmxServer.getAddress());
-    BeanQuery query =
-        BeanQuery.builder().domainName("java.lang").property("type", "Memory").build();
-
-    // act
-    client.close();
-
-    // assert
-    assertThrows(IllegalStateException.class, () -> client.queryBean(query));
-    assertThrows(
-        IllegalStateException.class, () -> client.queryBean(query, Collections.emptyList()));
-    assertThrows(IllegalStateException.class, () -> client.queryBeans(query));
-  }
-
   @Test
   void testCloseOnceMore() {
     // arrange
@@ -309,9 +293,18 @@ class MBeanClientTest {
   }
 
   @Test
-  void testUsePropertyListPattern() {
+  void testUsePropertyListPatternForRemote() {
+    testUsePropertyListPattern(MBeanClient.of(jmxServer.getAddress()));
+  }
+
+  @Test
+  void testUsePropertyListPatternForLocal() {
+    testUsePropertyListPattern(MBeanClient.local());
+  }
+
+  private void testUsePropertyListPattern(MBeanClient client) {
     // arrange
-    try (var client = MBeanClient.of(jmxServer.getAddress())) {
+    try (client) {
       BeanQuery patternQuery =
           BeanQuery.builder()
               .domainName("java.lang")
@@ -457,4 +450,10 @@ class MBeanClientTest {
       assertEquals(IntStream.range(0, 100).boxed().collect(toSet()), attrValues);
     }
   }
+
+  @Test
+  void testLocal() {
+    var client = MBeanClient.local();
+    Assertions.assertNotEquals(0, client.queryBeans(BeanQuery.all()).size());
+  }
 }