Repository: skiptests/astraea
Pull request #104 (2021-11-11 12:34:53)

Make sure test can run on windows
為了照顧windows使用者，因此這隻PR disable無法在windows上執行的tests，不過下列幾隻我不太確定為何不能在windows跑，因此會另外開議題來追蹤

1. `ReplicaCollieTest ` https://github.com/skiptests/astraea/issues/102
2. `TopicAdminTest#testReassign` https://github.com/skiptests/astraea/issues/103


Patch:
diff --git a/README.md b/README.md
index a14fced9..9a459681 100644
--- a/README.md
+++ b/README.md
@@ -154,25 +154,50 @@ This project offers a way to run kafka official tool by container. For example:
 
 ---
 
-## Kafka Metric Client
+## Kafka Metric Explorer
 
 This tool can be used to access Kafka's MBean metrics via JMX.
 
 Run the tool from source code
 
 ```shell
+# fetch every Mbeans from specific JMX server.
 ./gradlew run --args="metrics --jmx.server 192.168.50.178:1099"
+
+# fetch any Mbean that its object name contains property "type=Memory".
+./gradlew run --args="metrics --jmx.server 192.168.50.178:1099 --property type=Memory"
+
+# fetch any Mbean that belongs to "kafka.network" domain name, 
+# and it's object name contains two properties "request=Metadata" and "name=LocalTimeMs".
+./gradlew run --args="metrics --jmx.server 192.168.50.178:1099 --domain kafka.network --property request=Metadata --property name=LocalTimeMs"
+
+# list all Mbeans' object name on specific JMX server.
+./gradlew run --args="metrics --jmx.server 192.168.50.178:1099 --view-object-name-list"
 ```
 
 Run the tool from release
 ```shell
+# fetch every Mbeans from specific JMX server.
 java -jar app-0.0.1-SNAPSHOT-all.jar metrics --jmx.server 192.168.50.178:1099
+
+# fetch any Mbean that its object name contains property "type=Memory".
+java -jar app-0.0.1-SNAPSHOT-all.jar metrics --jmx.server 192.168.50.178:1099 --property type=Memory
+
+# fetch any Mbean that belongs to "kafka.network" domain name,
+# and it's object name contains two properties "request=Metadata" and "name=LocalTimeMs".
+java -jar app-0.0.1-SNAPSHOT-all.jar metrics --jmx.server 192.168.50.178:1099 --domain kafka.network --property request=Metadata --property name=LocalTimeMs
+
+# list all Mbeans' object name on specific JMX server.
+java -jar app-0.0.1-SNAPSHOT-all.jar metrics --jmx.server 192.168.50.178:1099 --view-object-name-list
 ```
 
-### Metric Client Configurations
+### Metric Explorer Configurations
 
-1. --jmx.server: the address to connect to Kafka JMX remote server
-2. --metrics: the Mbean metric to fetch. Default: All metrics
+1. --jmx.server: the address to connect to Kafka JMX remote server.
+2. --domain: query Mbeans from the specific domain name (support wildcard "\*" and "?"). Default: "\*".
+3. --property: query mbeans with the specific property (support wildcard "\*" and "?"). You can specify this argument multiple times. Default: [].
+4. --strict-match: only Mbeans with its object name completely match the given criteria shows. Default: false.
+5. --view-object-name-list: show the list view of MBeans' domain name & properties. Default: false.
 
 ---
 
diff --git a/app/src/main/java/org/astraea/App.java b/app/src/main/java/org/astraea/App.java
index 758a068a..a6ef0c43 100644
--- a/app/src/main/java/org/astraea/App.java
+++ b/app/src/main/java/org/astraea/App.java
@@ -5,7 +5,7 @@ import java.lang.reflect.InvocationTargetException;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
-import org.astraea.metrics.KafkaMetricClientApp;
+import org.astraea.metrics.MetricExplorer;
 import org.astraea.performance.Performance;
 import org.astraea.topic.ReplicaCollie;
 import org.astraea.topic.TopicExplorer;
@@ -14,7 +14,7 @@ public class App {
   private static final Map<String, Class<?>> MAIN_CLASSES =
       Map.of(
           "offset", TopicExplorer.class,
-          "metrics", KafkaMetricClientApp.class,
+          "metrics", MetricExplorer.class,
           "replica", ReplicaCollie.class,
           "performance", Performance.class);
 
diff --git a/app/src/main/java/org/astraea/metrics/KafkaMetricClientApp.java b/app/src/main/java/org/astraea/metrics/KafkaMetricClientApp.java
deleted file mode 100644
index 64ed43df..00000000
--- a/app/src/main/java/org/astraea/metrics/KafkaMetricClientApp.java
+++ /dev/null
@@ -1,88 +0,0 @@
-package org.astraea.metrics;
-
-import com.beust.jcommander.Parameter;
-import java.net.MalformedURLException;
-import java.time.LocalTime;
-import java.time.format.DateTimeFormatter;
-import java.time.format.FormatStyle;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-import javax.management.remote.JMXServiceURL;
-import org.astraea.argument.ArgumentUtil;
-import org.astraea.metrics.jmx.MBeanClient;
-import org.astraea.metrics.kafka.BrokerTopicMetricsResult;
-import org.astraea.metrics.kafka.KafkaMetrics;
-
-public final class KafkaMetricClientApp {
-
-  public static String JMX_URI_FORMAT = "service:jmx:rmi:///jndi/rmi://" + "%s" + "/jmxrmi";
-
-  public static String createJmxUrl(String address) {
-    return String.format(JMX_URI_FORMAT, address);
-  }
-
-  public static void main(String[] args) throws MalformedURLException {
-    var parameters = ArgumentUtil.parseArgument(new Argument(), args);
-
-    String argumentJmxServerNetworkAddress = parameters.address;
-    Set<String> argumentTargetMetrics = parameters.metrics;
-
-    JMXServiceURL serviceURL = new JMXServiceURL(createJmxUrl(argumentJmxServerNetworkAddress));
-    try (MBeanClient mBeanClient = new MBeanClient(serviceURL)) {
-
-      // find the actual metrics to fetch.
-      List<KafkaMetrics.BrokerTopic> metrics =
-          argumentTargetMetrics.stream()
-              .map(KafkaMetrics.BrokerTopic::of)
-              .collect(Collectors.toUnmodifiableList());
-
-      while (!Thread.interrupted()) {
-        // fetch
-        List<BrokerTopicMetricsResult> collect =
-            metrics.stream()
-                .map(x -> x.fetch(mBeanClient))
-                .collect(Collectors.toUnmodifiableList());
-
-        // output
-        System.out.println(
-            "["
-                + LocalTime.now().format(DateTimeFormatter.ofLocalizedTime(FormatStyle.MEDIUM))
-                + "]");
-        for (BrokerTopicMetricsResult result : collect) System.out.println(result);
-        System.out.println();
-
-        try {
-          TimeUnit.SECONDS.sleep(1);
-        } catch (InterruptedException e) {
-          break;
-        }
-      }
-
-    } catch (Exception e) {
-      e.printStackTrace();
-    }
-  }
-
-  static class Argument {
-    @Parameter(
-        names = {"--jmx.server"},
-        description = "The address to connect to JMX remote server",
-        validateWith = ArgumentUtil.NotEmptyString.class,
-        required = true)
-    String address;
-
-    @Parameter(
-        names = {"--metrics"},
-        variableArity = true,
-        converter = ArgumentUtil.StringSetConverter.class,
-        description =
-            "The metric names you want. If no metric name specified in argument, all metrics will be selected.")
-    Set<String> metrics =
-        Arrays.stream(KafkaMetrics.BrokerTopic.values())
-            .map(KafkaMetrics.BrokerTopic::metricName)
-            .collect(Collectors.toUnmodifiableSet());
-  }
-}
diff --git a/app/src/main/java/org/astraea/metrics/MetricExplorer.java b/app/src/main/java/org/astraea/metrics/MetricExplorer.java
new file mode 100644
index 00000000..26ac4cd1
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/MetricExplorer.java
@@ -0,0 +1,398 @@
+package org.astraea.metrics;
+
+import com.beust.jcommander.IParameterValidator;
+import com.beust.jcommander.IStringConverter;
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
+import java.net.MalformedURLException;
+import java.util.*;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import javax.management.openmbean.CompositeDataSupport;
+import javax.management.remote.JMXServiceURL;
+import org.astraea.argument.ArgumentUtil;
+import org.astraea.metrics.jmx.BeanObject;
+import org.astraea.metrics.jmx.BeanQuery;
+import org.astraea.metrics.jmx.MBeanClient;
+
+public class MetricExplorer {
+
+  static void execute(MBeanClient mBeanClient, Argument args) {
+    // compile query
+    BeanQuery.BeanQueryBuilder builder =
+        BeanQuery.builder(
+            args.fromDomainName,
+            args.properties.stream()
+                .map(Argument.CorrectPropertyFormat::toEntry)
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
+    if (!args.strictMatch) builder.usePropertyListPattern();
+
+    // execute query
+    Collection<BeanObject> beanObjects = mBeanClient.queryBeans(builder.build());
+
+    // display result
+    if (args.viewObjectNameList) displayObjectNameList(beanObjects);
+    else displayBeanObjects(beanObjects);
+  }
+
+  private static void displayBeanObjects(Collection<BeanObject> beanObjects) {
+    var propertyOrderMap = propertyOrder(beanObjects);
+    var collect = sortBeanObjects(beanObjects, propertyOrderMap);
+
+    for (BeanObject beanObject : collect) {
+      String properties =
+          beanObject.getProperties().entrySet().stream()
+              .sorted(Comparator.comparing(x -> propertyOrderMap.get(x.getKey())))
+              .map(entry -> entry.getKey() + "=" + entry.getValue())
+              .collect(Collectors.joining(","));
+
+      System.out.printf("[%s:%s]\n", beanObject.domainName(), properties);
+
+      Stream<String> strings = DataUtils.elaborateData(beanObject.getAttributes());
+      strings = DataUtils.streamAppendWith(" ", 4, strings);
+      strings.forEach(System.out::println);
+    }
+  }
+
+  private static void displayObjectNameList(Collection<BeanObject> beanObjects) {
+    var propertyOrderMap = propertyOrder(beanObjects);
+    var collect = sortBeanObjects(beanObjects, propertyOrderMap);
+
+    collect.forEach(
+        x ->
+            System.out.printf(
+                "%s:%s\n",
+                x.domainName(),
+                x.getProperties().entrySet().stream()
+                    .sorted(Comparator.comparingLong(z -> propertyOrderMap.get(z.getKey())))
+                    .map(z -> z.getKey() + "=" + z.getValue())
+                    .collect(Collectors.joining(","))));
+  }
+
+  /**
+   * generate the order of property entry, the lower number the higher priority
+   *
+   * @param beanObjects Mbeans objects, the order is decided by some property or statistics based on
+   *     the given Mbeans.
+   * @return a {@link Map} describe the priority of each property key
+   */
+  static Map<String, Long> propertyOrder(Collection<BeanObject> beanObjects) {
+
+    // count the frequency of each property key name
+    // property who has higher frequency will close to the tree root when printing result
+    Map<String, Long> propertyFrequencyMap =
+        beanObjects.stream()
+            .flatMap(x -> x.getProperties().entrySet().stream())
+            .map(Map.Entry::getKey)
+            .collect(Collectors.groupingBy(x -> x, Collectors.counting()));
+
+    // the order for sorting properties
+    Map<String, Long> propertyOrderMap =
+        propertyFrequencyMap.entrySet().stream()
+            .map(entry -> Map.entry(entry.getKey(), beanObjects.size() + 1 - entry.getValue()))
+            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+    // the property key "name" will always has the lowest priority, just like how JMC does it.
+    propertyOrderMap.put("name", beanObjects.size() + 1L);
+
+    return propertyOrderMap;
+  }
+
+  /**
+   * sort the given bean objects based on the given property order. The order is first decide by the
+   * alphabet order of domain name, then decided by the given property key order.
+   *
+   * @param beanObjects the Mbeans to be sorted.
+   * @param propertyOrderMap the order of property key, the lower number the higher priority.
+   * @return a new {@link List} consist of all the given Mbeans, all Mbeans are sort by given
+   *     property order.
+   */
+  static List<BeanObject> sortBeanObjects(
+      Collection<BeanObject> beanObjects, Map<String, Long> propertyOrderMap) {
+    return beanObjects.stream()
+        .sorted(
+            Comparator.comparing(BeanObject::domainName)
+                .thenComparing(
+                    (o1, o2) -> {
+                      List<Map.Entry<String, String>> order1 =
+                          o1.getProperties().entrySet().stream()
+                              .sorted(Comparator.comparing(x -> propertyOrderMap.get(x.getKey())))
+                              .collect(Collectors.toList());
+                      List<Map.Entry<String, String>> order2 =
+                          o2.getProperties().entrySet().stream()
+                              .sorted(Comparator.comparing(x -> propertyOrderMap.get(x.getKey())))
+                              .collect(Collectors.toList());
+
+                      // how the comparison works:
+                      // 1. sort all properties, now property key/value with higher key frequency
+                      // will show first.
+                      // 2. sort by property key/value amount
+                      // 3. compare the 1'st property key
+                      // 4. compare the 2'nd property key
+                      // 5. etc...
+                      // 6. compare the 1'st property value
+                      // 7. compare the 2'nd property value
+                      // 8. etc...
+                      // 9. object name equal
+
+                      if (order1.size() != order2.size()) {
+                        return -Integer.compare(order1.size(), order2.size());
+                      } else {
+                        var key =
+                            Comparator.comparing(
+                                (Map.Entry<String, String> x) -> propertyOrderMap.get(x.getKey()));
+                        var value = Map.Entry.<String, String>comparingByValue();
+
+                        for (int i = 0; i < order1.size(); i++) {
+                          int result = key.compare(order1.get(i), order2.get(i));
+                          if (result != 0) return result;
+                        }
+
+                        for (int i = 0; i < order1.size(); i++) {
+                          int result = value.compare(order1.get(i), order2.get(i));
+                          if (result != 0) return result;
+                        }
+
+                        return 0;
+                      }
+                    }))
+        .collect(Collectors.toList());
+  }
+
+  public static void main(String[] args) throws Exception {
+    var arguments = ArgumentUtil.parseArgument(new Argument(), args);
+
+    try (MBeanClient mBeanClient = new MBeanClient(arguments.jmxServer)) {
+      execute(mBeanClient, arguments);
+    }
+  }
+
+  static class Argument {
+    @Parameter(
+        names = {"--jmx.server"},
+        description =
+            "The JMX server address to connect to, support [hostname:port] style or JMX URI format",
+        validateWith = ArgumentUtil.NotEmptyString.class,
+        converter = Argument.JmxServerUrlConverter.class,
+        required = true)
+    JMXServiceURL jmxServer;
+
+    @Parameter(
+        names = {"--domain"},
+        description =
+            "Show Mbeans from specific domain pattern, support wildcard(*) and any char(?)")
+    String fromDomainName = "*";
+
+    @Parameter(
+        names = {"--property"},
+        description = "Fetch mbeans with specific property, support wildcard(*) and any char(?)",
+        validateWith = CorrectPropertyFormat.class)
+    List<String> properties = List.of();
+
+    @Parameter(
+        names = {"--strict-match"},
+        description =
+            "Only MBeans metrics with properties completely match the given criteria shows")
+    boolean strictMatch = false;
+
+    @Parameter(
+        names = {"--view-object-name-list"},
+        description = "Show the list view of MBeans' domain name & properties")
+    boolean viewObjectNameList = false;
+
+    public static class JmxServerUrlConverter implements IStringConverter<JMXServiceURL> {
+
+      /** This regex used to test if a string look like a JMX URL */
+      static Pattern patternOfJmxUrlStart = Pattern.compile("^service:jmx:rmi://");
+
+      /**
+       * Convert the given string to a {@link JMXServiceURL}.
+       *
+       * <p>Following string formats are supported
+       *
+       * <ol>
+       *   <li>From hostname and port {@code "kafka1.example.com:9875"}
+       *   <li>From JMX Service URL string {@code "service:jmx:rmi:///kafka1.example.come ..."}
+       * </ol>
+       */
+      @Override
+      public JMXServiceURL convert(String value) {
+        Matcher matchJmxUrlStart = patternOfJmxUrlStart.matcher(value);
+
+        if (matchJmxUrlStart.find()) {
+          return convertFromJmxUrl(value);
+        } else {
+          return convertFromAddress(value);
+        }
+      }
+
+      private JMXServiceURL convertFromJmxUrl(String jmxUrlString) {
+        try {
+          return new JMXServiceURL(jmxUrlString);
+        } catch (MalformedURLException e) {
+          throw new IllegalArgumentException("Cannot parse given URL: " + jmxUrlString, e);
+        }
+      }
+
+      private JMXServiceURL convertFromAddress(String addressOfJmxServer) {
+        String url = String.format("service:jmx:rmi:///jndi/rmi://%s/jmxrmi", addressOfJmxServer);
+        try {
+          return new JMXServiceURL(url);
+        } catch (MalformedURLException e) {
+          throw new IllegalArgumentException("Illegal JMX Url: " + url, e);
+        }
+      }
+    }
+
+    public static class CorrectPropertyFormat implements IParameterValidator {
+
+      static Pattern propertyPattern = Pattern.compile("^(?<key>[^=]+)=(?<value>.+)$");
+
+      @Override
+      public void validate(String name, String value) throws ParameterException {
+        if (!propertyPattern.matcher(value).find())
+          throw new ParameterException(
+              name
+                  + "with value ["
+                  + value
+                  + "] should match format: "
+                  + propertyPattern.pattern());
+      }
+
+      static Map.Entry<String, String> toEntry(String property) {
+        Matcher matcher = propertyPattern.matcher(property);
+        if (!matcher.find()) throw new IllegalArgumentException("Not property format: " + property);
+        return Map.entry(matcher.group("key"), matcher.group("value"));
+      }
+    }
+  }
+
+  static class DataUtils {
+
+    static int stringLengthLimit = 500;
+
+    static Stream<String> elaborateData(Object target) {
+      try {
+        if (target == null) {
+          return Stream.of("null");
+        } else if (target instanceof List<?>) {
+          return elaborateList((List<?>) target);
+        } else if (target instanceof Map<?, ?>) {
+          return elaborateMap((Map<?, ?>) target);
+        } else if (target instanceof CompositeDataSupport) {
+          return elaborateCompositeDataSupport(((CompositeDataSupport) target));
+        } else if (target.getClass().isArray()) {
+          if (target.getClass() == long[].class) return elaborateArray((long[]) target);
+          else return elaborateArray((Object[]) target);
+        }
+
+        // I have no idea how to display this object, just toString it.
+        return Stream.of(stringify(target));
+      } catch (Exception e) {
+        return Stream.of(e.toString());
+      }
+    }
+
+    static String stringify(Object object) {
+      String string = object.toString();
+      if (string.length() > stringLengthLimit) {
+        return string.substring(0, stringLengthLimit)
+            + String.format("%n... (%d characters truncated)", string.length() - stringLengthLimit);
+      } else {
+        return string;
+      }
+    }
+
+    static Stream<String> elaborateMap(Map<?, ?> data) {
+      try {
+        return data.entrySet().stream()
+            .sorted(Comparator.comparing(Object::toString))
+            .flatMap(DataUtils::elaborateMapEntry);
+      } catch (Exception e) {
+        return Stream.of(e.toString());
+      }
+    }
+
+    static Stream<String> elaborateMapEntry(Map.Entry<?, ?> entry) {
+      try {
+        List<String> key = elaborateData(entry.getKey()).collect(Collectors.toList());
+        List<String> value = elaborateData(entry.getValue()).collect(Collectors.toList());
+
+        if (key.size() == 1 && value.size() > 1)
+          return Stream.concat(
+              Stream.of(String.format("\"%s\":", key.get(0))),
+              streamAppendWith(" ", 4, value.stream()));
+
+        return Stream.of(String.format("\"%s\": %s", key.get(0), value.get(0)));
+      } catch (Exception e) {
+        return Stream.of(e.toString());
+      }
+    }
+
+    static Stream<String> elaborateArray(Object[] array) {
+      try {
+        return IntStream.range(0, array.length)
+            .boxed()
+            .flatMap(
+                i -> {
+                  List<String> collect = elaborateData(array[i]).collect(Collectors.toList());
+                  if (collect.size() == 1)
+                    return Stream.of(String.format("%d: %s", i, collect.get(0)));
+                  else
+                    return Stream.concat(
+                        Stream.of(i + ":"), streamAppendWith(" ", 4, collect.stream()));
+                });
+      } catch (Exception e) {
+        return Stream.of(e.toString());
+      }
+    }
+
+    static Stream<String> elaborateArray(long[] array) {
+      try {
+        return IntStream.range(0, array.length).mapToObj(x -> x + ": " + array[x]);
+      } catch (Exception e) {
+        return Stream.of(e.toString());
+      }
+    }
+
+    static Stream<String> elaborateList(List<?> list) {
+      try {
+        return IntStream.range(0, list.size())
+            .boxed()
+            .flatMap(
+                i -> {
+                  List<String> collect = elaborateData(list.get(i)).collect(Collectors.toList());
+                  if (collect.size() == 1)
+                    return Stream.of(String.format("%d: %s", i, collect.get(0)));
+                  else
+                    return Stream.concat(
+                        Stream.of(i + ":"), streamAppendWith(" ", 4, collect.stream()));
+                });
+      } catch (Exception e) {
+        return Stream.of(e.toString());
+      }
+    }
+
+    static Stream<String> elaborateCompositeDataSupport(CompositeDataSupport data) {
+      try {
+        return data.getCompositeType().keySet().stream()
+            .sorted()
+            .flatMap(key -> DataUtils.elaborateMapEntry(Map.entry(key, data.get(key))));
+      } catch (Exception e) {
+        return Stream.of(e.toString());
+      }
+    }
+
+    static Stream<String> streamAppendWith(String s, int size, Stream<String> source) {
+      try {
+        String append = String.join("", Collections.nCopies(size, s));
+        return source.map(x -> append + x);
+      } catch (Exception e) {
+        return Stream.of(e.toString());
+      }
+    }
+  }
+}


Test patch:
diff --git a/app/src/test/java/org/astraea/metrics/MetricExplorerTest.java b/app/src/test/java/org/astraea/metrics/MetricExplorerTest.java
new file mode 100644
index 00000000..7aa324c6
--- /dev/null
+++ b/app/src/test/java/org/astraea/metrics/MetricExplorerTest.java
@@ -0,0 +1,123 @@
+package org.astraea.metrics;
+
+import static org.astraea.metrics.MetricExplorer.*;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import org.astraea.argument.ArgumentUtil;
+import org.astraea.metrics.jmx.BeanObject;
+import org.astraea.metrics.jmx.MBeanClient;
+import org.junit.jupiter.api.function.Executable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class MetricExplorerTest {
+
+  @ParameterizedTest
+  @CsvSource(
+      delimiterString = "->",
+      value = {
+        "service:jmx:rmi://localhost:1234/jndi/rmi://localhost:2344/jmxrmi     -> match",
+        "service:jmx:rmi:///jndi/rmi://:9999/jmxrmi                            -> match",
+        "service:jmx:rmi://myhost/jndi/rmi://myhost:1099/myhost/myjmxconnector -> match",
+        "service:jmx:rmi:///jndi/rmi://localhost:9987/jmxrmi                   -> match",
+        "aaaaaaaaaaaaaaaaaaa                                                   -> no-match",
+        "bbbbbbbbbbbbbbbbbbb                                                   -> no-match",
+      })
+  void testPatternJmxUrlStart(String url, String expected) {
+    // arrange
+    Pattern pattern = Argument.JmxServerUrlConverter.patternOfJmxUrlStart;
+
+    // act
+    Matcher matcher = pattern.matcher(url);
+
+    // assert
+    assertEquals(expected, matcher.find() ? "match" : "no-match");
+  }
+
+  @ParameterizedTest
+  @CsvSource(
+      delimiterString = "->",
+      value = {
+        "type=Memory                      -> type            -> Memory",
+        "type=OperatingSystem             -> type            -> OperatingSystem",
+        "key=value                        -> key             -> value",
+        "logDirectory=\"/tmp/kafka-logs\" -> logDirectory    -> \"/tmp/kafka-logs\"",
+        "face=\"= =\"                     -> face            -> \"= =\"",
+      })
+  void testPatternProperty(String property, String key, String value) {
+    // arrange
+    Pattern pattern = Argument.CorrectPropertyFormat.propertyPattern;
+
+    // act
+    Matcher matcher = pattern.matcher(property);
+
+    // assert
+    assertTrue(matcher.find());
+    assertEquals(key, matcher.group("key"));
+    assertEquals(value, matcher.group("value"));
+  }
+
+  @ParameterizedTest
+  @ValueSource(
+      strings = {
+        "--jmx.server localhost:5566 --domain example.com",
+        "--jmx.server localhost:5566 --domain example.com --view-object-name-list",
+      })
+  void executeDoesPrintSomething(String args) {
+    // arrange
+    var argument = ArgumentUtil.parseArgument(new Argument(), args.split(" "));
+    var mockMBeanClient = mock(MBeanClient.class);
+    when(mockMBeanClient.queryBeans(any()))
+        .thenReturn(List.of(new BeanObject("example.com", Map.of("key", "value"), Map.of())));
+    var stdout = System.out;
+    var mockOutput = new ByteArrayOutputStream();
+    System.setOut(new PrintStream(mockOutput));
+
+    // act
+    Executable executable = () -> MetricExplorer.execute(mockMBeanClient, argument);
+
+    // assert
+    assertDoesNotThrow(executable);
+    assertTrue(mockOutput.toString().contains("example.com"));
+    assertTrue(mockOutput.toString().contains("key"));
+    assertTrue(mockOutput.toString().contains("value"));
+
+    // restore
+    System.setOut(stdout);
+  }
+
+  @ParameterizedTest
+  @CsvSource(
+      delimiterString = "(is",
+      value = {
+        "--jmx.server localhost:5566                                       (is ok",
+        "--jmx.server localhost:5566 --domain kafka.log                    (is ok",
+        "--jmx.server localhost:5566 --property type=Memory                (is ok",
+        "--jmx.server localhost:5566 --strict-match --property type=Memory (is ok",
+        "--jmx.server localhost:5566 --view-object-name-list               (is ok",
+        "wuewuewuewue                                                      (is not ok",
+        "--view-object-name-list                                           (is not ok",
+      })
+  void ensureArgumentWorking(String argumentString, String outcome) {
+    // arrange
+    String[] arguments = argumentString.split(" ");
+
+    // act
+    Executable doParsing = () -> ArgumentUtil.parseArgument(new Argument(), arguments);
+
+    // assert
+    if (outcome.equals("ok")) {
+      assertDoesNotThrow(doParsing);
+    } else if (outcome.equals("not ok")) {
+      assertThrows(Exception.class, doParsing);
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientAppTest.java b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientAppTest.java
deleted file mode 100644
index a8d7ac5c..00000000
--- a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientAppTest.java
+++ /dev/null
@@ -1,33 +0,0 @@
-package org.astraea.metrics.kafka;
-
-import static org.junit.jupiter.api.Assertions.*;
-
-import com.beust.jcommander.ParameterException;
-import java.io.OutputStream;
-import java.io.PrintStream;
-import org.astraea.metrics.KafkaMetricClientApp;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-class KafkaMetricClientAppTest {
-
-  private final PrintStream stderr = System.err;
-
-  @BeforeEach
-  void beforeEach() {
-    // swallow stderr output to null stream, avoid help() info from KafkaMetricClientApp
-    System.setErr(new PrintStream(OutputStream.nullOutputStream()));
-  }
-
-  @AfterEach
-  void afterEach() {
-    // restore stderr stream
-    System.setErr(stderr);
-  }
-
-  @Test
-  void testBadArguments() {
-    assertThrows(ParameterException.class, () -> KafkaMetricClientApp.main(new String[0]));
-  }
-}
diff --git a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java
index a9ecc750..794dd3ad 100644
--- a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java
+++ b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java
@@ -1,6 +1,8 @@
 package org.astraea.metrics.kafka;
 
 import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.condition.OS.LINUX;
+import static org.junit.jupiter.api.condition.OS.WINDOWS;
 
 import java.io.IOException;
 import java.lang.management.ManagementFactory;
@@ -19,6 +21,8 @@ import org.astraea.topic.TopicAdmin;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.EnabledOnOs;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
 
@@ -145,8 +149,6 @@ class KafkaMetricsTest extends RequireBrokerCluster {
     assertDoesNotThrow(operatingSystemInfo::committedVirtualMemorySize);
     assertDoesNotThrow(operatingSystemInfo::freePhysicalMemorySize);
     assertDoesNotThrow(operatingSystemInfo::freeSwapSpaceSize);
-    assertDoesNotThrow(operatingSystemInfo::maxFileDescriptorCount);
-    assertDoesNotThrow(operatingSystemInfo::openFileDescriptorCount);
     assertDoesNotThrow(operatingSystemInfo::name);
     assertDoesNotThrow(operatingSystemInfo::processCpuLoad);
     assertDoesNotThrow(operatingSystemInfo::processCpuTime);
@@ -157,6 +159,20 @@ class KafkaMetricsTest extends RequireBrokerCluster {
     assertDoesNotThrow(operatingSystemInfo::version);
   }
 
+  @Test
+  @DisabledOnOs(WINDOWS)
+  void maxFileDescriptorCount() {
+    OperatingSystemInfo operatingSystemInfo = KafkaMetrics.Host.operatingSystem(mBeanClient);
+    assertDoesNotThrow(operatingSystemInfo::maxFileDescriptorCount);
+  }
+
+  @Test
+  @DisabledOnOs(WINDOWS)
+  void openFileDescriptorCount() {
+    OperatingSystemInfo operatingSystemInfo = KafkaMetrics.Host.operatingSystem(mBeanClient);
+    assertDoesNotThrow(operatingSystemInfo::openFileDescriptorCount);
+  }
+
   @Test
   void testJvmMemory() {
     JvmMemory jvmMemory = KafkaMetrics.Host.jvmMemory(mBeanClient);
@@ -172,11 +188,13 @@ class KafkaMetricsTest extends RequireBrokerCluster {
   }
 
   @Test
+  @EnabledOnOs(LINUX)
   void linuxDiskReadBytes() {
     assertDoesNotThrow(() -> KafkaMetrics.BrokerTopic.linuxDiskReadBytes(mBeanClient));
   }
 
   @Test
+  @EnabledOnOs(LINUX)
   void linuxDiskWriteBytes() {
     assertDoesNotThrow(() -> KafkaMetrics.BrokerTopic.linuxDiskWriteBytes(mBeanClient));
   }
diff --git a/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java b/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java
index 013d96bd..14008443 100644
--- a/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java
+++ b/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java
@@ -1,5 +1,7 @@
 package org.astraea.topic;
 
+import static org.junit.jupiter.api.condition.OS.WINDOWS;
+
 import java.io.IOException;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
@@ -9,15 +11,18 @@ import org.astraea.Utils;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
 
 public class ReplicaCollieTest extends RequireBrokerCluster {
 
   @Test
+  @DisabledOnOs(WINDOWS)
   void testVerify() throws IOException, InterruptedException {
     test(true);
   }
 
   @Test
+  @DisabledOnOs(WINDOWS)
   void testExecute() throws IOException, InterruptedException {
     test(false);
   }
diff --git a/app/src/test/java/org/astraea/topic/TopicAdminTest.java b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
index f94f4b50..4f5e6d7c 100644
--- a/app/src/test/java/org/astraea/topic/TopicAdminTest.java
+++ b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
@@ -1,5 +1,7 @@
 package org.astraea.topic;
 
+import static org.junit.jupiter.api.condition.OS.WINDOWS;
+
 import java.io.IOException;
 import java.util.Collection;
 import java.util.Map;
@@ -14,6 +16,7 @@ import org.astraea.producer.Producer;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
 
 public class TopicAdminTest extends RequireBrokerCluster {
 
@@ -132,6 +135,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
   }
 
   @Test
+  @DisabledOnOs(WINDOWS)
   void testReassign() throws IOException, InterruptedException {
     var topicName = "testReassign";
     try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {