Repository: skiptests/astraea
Pull request #36 (2021-10-02 05:35:49)

Add outline of tools
在readme前面增加一個目錄來概括有哪些工具可用


Patch:
diff --git a/README.md b/README.md
index 9a949afd..38e568c5 100644
--- a/README.md
+++ b/README.md
@@ -12,6 +12,11 @@ a collection of tools used to balance Kafka data
 
 This project offers many kafka tools to simplify the life for kafka users.
 
+1. [Kafka quick start](#kafka-cluster-quick-start): set up a true kafka cluster in one minute
+2. [Kafka benchmark](#latency-benchmark): run producers/consumers to test the performance and consistency for kafka cluster
+3. [Kafka offset explorer](#offset-explorer): check the start/end offsets of kafka topics
+4. [Kafka official tool](#kafka-official-tool): run any one specific kafka official tool. All you have to prepare is the docker env.
+
 ---
 
 ## Kafka Cluster Quick Start
diff --git a/app/src/main/java/org/astraea/App.java b/app/src/main/java/org/astraea/App.java
index a6ad139e..c0db58cf 100644
--- a/app/src/main/java/org/astraea/App.java
+++ b/app/src/main/java/org/astraea/App.java
@@ -4,12 +4,13 @@ import java.util.Arrays;
 import java.util.List;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
+import org.astraea.metrics.kafka.KafkaMetricClientApp;
 import org.astraea.offset.OffsetExplorer;
 import org.astraea.performance.latency.End2EndLatency;
 
 public class App {
   private static final List<Class<?>> MAIN_CLASSES =
-      Arrays.asList(End2EndLatency.class, OffsetExplorer.class);
+      Arrays.asList(End2EndLatency.class, OffsetExplorer.class, KafkaMetricClientApp.class);
 
   private static String toString(List<Class<?>> mains) {
     return mains.stream().map(Class::getName).collect(Collectors.joining(","));
diff --git a/app/src/main/java/org/astraea/metrics/kafka/KafkaMetricClient.java b/app/src/main/java/org/astraea/metrics/kafka/KafkaMetricClient.java
new file mode 100644
index 00000000..48a9655b
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/KafkaMetricClient.java
@@ -0,0 +1,49 @@
+package org.astraea.metrics.kafka;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import javax.management.remote.JMXServiceURL;
+import org.astraea.metrics.jmx.BeanObject;
+import org.astraea.metrics.jmx.BeanQuery;
+import org.astraea.metrics.jmx.MBeanClient;
+import org.astraea.metrics.kafka.metrics.BrokerTopicMetrics;
+import org.astraea.metrics.kafka.metrics.BrokerTopicMetricsResult;
+
+public class KafkaMetricClient implements AutoCloseable {
+
+  private final MBeanClient mBeanClient;
+
+  public KafkaMetricClient(JMXServiceURL serviceURL) {
+    try {
+      this.mBeanClient = new MBeanClient(serviceURL);
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
+  /**
+   * Request BrokerTopicMetrics.
+   *
+   * <p>For the available metrics, checkout {@link BrokerTopicMetrics}.
+   *
+   * @param metric the metric to request
+   * @return a {@link BrokerTopicMetricsResult} indicate the result
+   */
+  public BrokerTopicMetricsResult requestMetric(BrokerTopicMetrics metric) {
+    BeanObject beanObject =
+        mBeanClient
+            .tryQueryBean(
+                BeanQuery.builder("kafka.server")
+                    .property("type", "BrokerTopicMetrics")
+                    .property("name", metric.metricName())
+                    .build())
+            .orElseThrow();
+
+    return new BrokerTopicMetricsResult(metric, beanObject);
+  }
+
+  @Override
+  public void close() throws Exception {
+    this.mBeanClient.close();
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/kafka/KafkaMetricClientApp.java b/app/src/main/java/org/astraea/metrics/kafka/KafkaMetricClientApp.java
new file mode 100644
index 00000000..33cfeb48
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/KafkaMetricClientApp.java
@@ -0,0 +1,87 @@
+package org.astraea.metrics.kafka;
+
+import java.net.MalformedURLException;
+import java.time.LocalTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.FormatStyle;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import javax.management.remote.JMXServiceURL;
+import org.astraea.metrics.kafka.metrics.BrokerTopicMetrics;
+import org.astraea.metrics.kafka.metrics.BrokerTopicMetricsResult;
+
+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 {
+
+    // ensure argument safe
+    if (args.length < 1) {
+      help();
+      throw new IllegalArgumentException();
+    }
+
+    String argumentJmxServerNetworkAddress = args[0];
+    List<String> argumentTargetMetrics = List.of(args).subList(1, args.length);
+
+    JMXServiceURL serviceURL = new JMXServiceURL(createJmxUrl(argumentJmxServerNetworkAddress));
+    try (KafkaMetricClient kafkaMetricClient = new KafkaMetricClient(serviceURL)) {
+
+      // find the actual metrics to fetch.
+      List<BrokerTopicMetrics> metrics =
+          argumentTargetMetrics.stream()
+              .map(BrokerTopicMetrics::valueOf)
+              .collect(Collectors.toUnmodifiableList());
+
+      // if no metric name specified, all metrics are selected
+      if (argumentTargetMetrics.size() == 0) metrics = List.of(BrokerTopicMetrics.values());
+
+      while (!Thread.interrupted()) {
+        // fetch
+        List<BrokerTopicMetricsResult> collect =
+            metrics.stream()
+                .map(kafkaMetricClient::requestMetric)
+                .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();
+      help();
+    }
+  }
+
+  private static void help() {
+    String simpleName = KafkaMetricClientApp.class.getSimpleName();
+    System.err.printf("Usage: %s <jmx server address> [metric names ...]\n", simpleName);
+    System.err.println();
+    System.err.println("If no metric name specified in argument, all metrics will be selected.");
+    System.err.println();
+    System.err.printf("Example 1: %s localhost:9875\n", simpleName);
+    System.err.printf("Example 2: %s localhost:9875 BytesInPerSec BytesOutPerSec\n", simpleName);
+    System.err.println();
+    System.err.println("Available Metrics:");
+    for (BrokerTopicMetrics value : BrokerTopicMetrics.values()) {
+      System.err.printf("    %s\n", value.metricName());
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetrics.java b/app/src/main/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetrics.java
new file mode 100644
index 00000000..a772609e
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetrics.java
@@ -0,0 +1,60 @@
+package org.astraea.metrics.kafka.metrics;
+
+public enum BrokerTopicMetrics {
+  /** Message validation failure rate due to non-continuous offset or sequence number in batch */
+  InvalidOffsetOrSequenceRecordsPerSec("InvalidOffsetOrSequenceRecordsPerSec"),
+
+  /** Message validation failure rate due to incorrect crc checksum */
+  InvalidMessageCrcRecordsPerSec("InvalidMessageCrcRecordsPerSec"),
+
+  FetchMessageConversionsPerSec("FetchMessageConversionsPerSec"),
+
+  BytesRejectedPerSec("BytesRejectedPerSec"),
+
+  /** Message in rate */
+  MessagesInPerSec("MessagesInPerSec"),
+
+  /** Incoming byte rate of reassignment traffic */
+  ReassignmentBytesInPerSec("ReassignmentBytesInPerSec"),
+
+  FailedFetchRequestsPerSec("FailedFetchRequestsPerSec"),
+
+  /** Byte in rate from other brokers */
+  ReplicationBytesInPerSec("ReplicationBytesInPerSec"),
+
+  /** Message validation failure rate due to no key specified for compacted topic */
+  NoKeyCompactedTopicRecordsPerSec("NoKeyCompactedTopicRecordsPerSec"),
+
+  TotalFetchRequestsPerSec("TotalFetchRequestsPerSec"),
+
+  FailedProduceRequestsPerSec("FailedProduceRequestsPerSec"),
+
+  /** Byte in rate from clients */
+  BytesInPerSec("BytesInPerSec"),
+
+  TotalProduceRequestsPerSec("TotalProduceRequestsPerSec"),
+
+  /** Message validation failure rate due to invalid magic number */
+  InvalidMagicNumberRecordsPerSec("InvalidMagicNumberRecordsPerSec"),
+
+  /** Outgoing byte rate of reassignment traffic */
+  ReassignmentBytesOutPerSec("ReassignmentBytesOutPerSec"),
+
+  /** Bytes in rate from other brokers */
+  ReplicationBytesOutPerSec("ReplicationBytesOutPerSec"),
+
+  ProduceMessageConversionsPerSec("ProduceMessageConversionsPerSec"),
+
+  /** Byte out rate to clients. */
+  BytesOutPerSec("BytesOutPerSec");
+
+  private final String metricName;
+
+  BrokerTopicMetrics(String name) {
+    this.metricName = name;
+  }
+
+  public String metricName() {
+    return metricName;
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetricsResult.java b/app/src/main/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetricsResult.java
new file mode 100644
index 00000000..f3c77410
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetricsResult.java
@@ -0,0 +1,35 @@
+package org.astraea.metrics.kafka.metrics;
+
+import java.util.Map;
+import java.util.Objects;
+import org.astraea.metrics.jmx.BeanObject;
+import org.astraea.metrics.kafka.metrics.modifiers.*;
+
+public class BrokerTopicMetricsResult implements HasCount, HasEventType, HasRate {
+
+  private final BeanObject beanObject;
+  private final BrokerTopicMetrics metric;
+
+  public BrokerTopicMetricsResult(BrokerTopicMetrics metric, BeanObject beanObject) {
+    this.beanObject = Objects.requireNonNull(beanObject);
+    this.metric = Objects.requireNonNull(metric);
+  }
+
+  @Override
+  public String toString() {
+    StringBuilder sb = new StringBuilder();
+    for (Map.Entry<String, Object> e : beanObject().getAttributes().entrySet()) {
+      sb.append(System.lineSeparator())
+          .append("  ")
+          .append(e.getKey())
+          .append("=")
+          .append(e.getValue());
+    }
+    return metric.name() + "{" + sb + "}";
+  }
+
+  @Override
+  public BeanObject beanObject() {
+    return beanObject;
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasBeanObject.java b/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasBeanObject.java
new file mode 100644
index 00000000..608e6aff
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasBeanObject.java
@@ -0,0 +1,7 @@
+package org.astraea.metrics.kafka.metrics.modifiers;
+
+import org.astraea.metrics.jmx.BeanObject;
+
+public interface HasBeanObject {
+  BeanObject beanObject();
+}
diff --git a/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasCount.java b/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasCount.java
new file mode 100644
index 00000000..236bbfff
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasCount.java
@@ -0,0 +1,7 @@
+package org.astraea.metrics.kafka.metrics.modifiers;
+
+public interface HasCount extends HasBeanObject {
+  default long count() {
+    return (long) beanObject().getAttributes().getOrDefault("Count", 0);
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasEventType.java b/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasEventType.java
new file mode 100644
index 00000000..827209a4
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasEventType.java
@@ -0,0 +1,7 @@
+package org.astraea.metrics.kafka.metrics.modifiers;
+
+public interface HasEventType extends HasBeanObject {
+  default String eventType() {
+    return (String) beanObject().getAttributes().get("EventType");
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasRate.java b/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasRate.java
new file mode 100644
index 00000000..91627ae1
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasRate.java
@@ -0,0 +1,25 @@
+package org.astraea.metrics.kafka.metrics.modifiers;
+
+import java.util.concurrent.TimeUnit;
+
+public interface HasRate extends HasBeanObject {
+  default double meanRate() {
+    return (double) beanObject().getAttributes().getOrDefault("MeanRate", 0);
+  }
+
+  default double oneMinuteRate() {
+    return (double) beanObject().getAttributes().getOrDefault("OneMinuteRate", 0);
+  }
+
+  default double fiveMinuteRate() {
+    return (double) beanObject().getAttributes().getOrDefault("FiveMinuteRate", 0);
+  }
+
+  default double fifteenMinuteRate() {
+    return (double) beanObject().getAttributes().getOrDefault("FifteenMinuteRate", 0);
+  }
+
+  default TimeUnit rateUnit() {
+    return (TimeUnit) beanObject().getAttributes().get("RateUnit");
+  }
+}


Test patch:
diff --git a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientAppTest.java b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientAppTest.java
new file mode 100644
index 00000000..6d2db880
--- /dev/null
+++ b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientAppTest.java
@@ -0,0 +1,31 @@
+package org.astraea.metrics.kafka;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.io.OutputStream;
+import java.io.PrintStream;
+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(IllegalArgumentException.class, () -> KafkaMetricClientApp.main(new String[0]));
+  }
+}
diff --git a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientTest.java b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientTest.java
new file mode 100644
index 00000000..d4c58782
--- /dev/null
+++ b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientTest.java
@@ -0,0 +1,103 @@
+package org.astraea.metrics.kafka;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import javax.management.*;
+import javax.management.remote.JMXConnectorServer;
+import javax.management.remote.JMXConnectorServerFactory;
+import javax.management.remote.JMXServiceURL;
+import org.astraea.metrics.jmx.Utility;
+import org.astraea.metrics.kafka.metrics.BrokerTopicMetrics;
+import org.astraea.metrics.kafka.metrics.BrokerTopicMetricsResult;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+class KafkaMetricClientTest {
+
+  private MBeanServer mBeanServer;
+  private JMXConnectorServer jmxServer;
+  private Map<ObjectName, Object> registeredBeans = new HashMap<>();
+  private KafkaMetricClient sut;
+
+  private void register(ObjectName name, Object mBean) {
+    registeredBeans.put(name, mBean);
+    try {
+      mBeanServer.registerMBean(mBean, name);
+    } catch (InstanceAlreadyExistsException
+        | MBeanRegistrationException
+        | NotCompliantMBeanException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private void clearRegisteredMBeans() {
+    registeredBeans.forEach(
+        (name, mbeans) -> {
+          try {
+            mBeanServer.unregisterMBean(name);
+          } catch (InstanceNotFoundException | MBeanRegistrationException e) {
+            throw new RuntimeException(e);
+          }
+        });
+    registeredBeans.clear();
+  }
+
+  @BeforeEach
+  void setUp() throws IOException {
+    JMXServiceURL serviceURL = new JMXServiceURL("service:jmx:rmi://127.0.0.1");
+
+    mBeanServer = ManagementFactory.getPlatformMBeanServer();
+
+    jmxServer = JMXConnectorServerFactory.newJMXConnectorServer(serviceURL, null, mBeanServer);
+    jmxServer.start();
+
+    sut = new KafkaMetricClient(jmxServer.getAddress());
+  }
+
+  @AfterEach
+  void tearDown() throws Exception {
+    jmxServer.stop();
+    clearRegisteredMBeans();
+    mBeanServer = null;
+    sut.close();
+  }
+
+  @ParameterizedTest
+  @EnumSource(value = BrokerTopicMetrics.class)
+  void testRequestBrokerTopicMetrics(BrokerTopicMetrics metric)
+      throws MalformedObjectNameException {
+    // arrange
+    Object mbean =
+        Utility.createReadOnlyDynamicMBean(
+            Map.of(
+                "Count", 1L,
+                "EventType", "bytes",
+                "FifteenMinuteRate", 2.0,
+                "FiveMinuteRate", 4.0,
+                "MeanRate", 8.0,
+                "OneMinuteRate", 16.0,
+                "RateUnit", TimeUnit.SECONDS));
+    register(
+        ObjectName.getInstance("kafka.server:type=BrokerTopicMetrics,name=" + metric.name()),
+        mbean);
+
+    // act
+    BrokerTopicMetricsResult result = sut.requestMetric(metric);
+
+    // assert
+    assertEquals(1L, result.count());
+    assertEquals("bytes", result.eventType());
+    assertEquals(2.0, result.fifteenMinuteRate());
+    assertEquals(4.0, result.fiveMinuteRate());
+    assertEquals(8.0, result.meanRate());
+    assertEquals(16.0, result.oneMinuteRate());
+    assertEquals(TimeUnit.SECONDS, result.rateUnit());
+  }
+}
diff --git a/app/src/test/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetricsTest.java b/app/src/test/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetricsTest.java
new file mode 100644
index 00000000..a38c37d7
--- /dev/null
+++ b/app/src/test/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetricsTest.java
@@ -0,0 +1,23 @@
+package org.astraea.metrics.kafka.metrics;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.util.Arrays;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.junit.jupiter.api.Test;
+
+class BrokerTopicMetricsTest {
+
+  @Test
+  void testAllEnumNameUnique() {
+    // arrange act
+    Set<String> collectedName =
+        Arrays.stream(BrokerTopicMetrics.values())
+            .map(BrokerTopicMetrics::metricName)
+            .collect(Collectors.toSet());
+
+    // assert
+    assertEquals(BrokerTopicMetrics.values().length, collectedName.size());
+  }
+}