Repository: skiptests/astraea
Pull request #335 (2022-05-13 16:55:21)

Enable to expose broker information by Restful API
```
GET http://hostname:port/brokers
```

```json
{
  "brokers": [
    {
      "id": 1001,
      "configs": {
        "log.cleaner.min.compaction.lag.ms": "0",
        "metric.reporters": "",
        "offsets.topic.num.partitions": "50",
        "sasl.oauthbearer.jwks.endpoint.refresh.ms": "3600000",
        "log.flush.interval.messages": "9223372036854775807",
        "controller.socket.timeout.ms": "30000",
        "auto.create.topics.enable": "true",
        "principal.builder.class": "org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder",
        "controller.quorum.request.timeout.ms": "2000",
        "replica.socket.receive.buffer.bytes": "65536",
        "min.insync.replicas": "1",
        "replica.fetch.wait.max.ms": "500",
        "num.recovery.threads.per.data.dir": "1",
        "ssl.keystore.type": "JKS",
        "password.encoder.iterations": "4096",
        "zookeeper.ssl.protocol": "TLSv1.2",
        "sasl.mechanism.inter.broker.protocol": "GSSAPI",
        "default.replication.factor": "1",
        "sasl.kerberos.principal.to.local.rules": "DEFAULT",
        "log.preallocate": "false",
        "metadata.log.segment.bytes": "1073741824",
        "fetch.purgatory.purge.interval.requests": "1000",
        "ssl.endpoint.identification.algorithm": "https",
        "replica.socket.timeout.ms": "30000",
        "message.max.bytes": "1048588",
        "transactional.id.expiration.ms": "604800000",
        "transaction.state.log.replication.factor": "1",
        "num.io.threads": "8",
        "sasl.login.refresh.buffer.seconds": "300",
        "max.connection.creation.rate": "2147483647",
        "connections.max.reauth.ms": "0",
        "offsets.commit.required.acks": "-1",
        "connection.failed.authentication.delay.ms": "100",
        "log.flush.offset.checkpoint.interval.ms": "60000",
        "delete.topic.enable": "true",
        "quota.window.size.seconds": "1",
        "zookeeper.ssl.client.enable": "false",
        "ssl.truststore.type": "JKS",
        "offsets.commit.timeout.ms": "5000",
        "quota.window.num": "11",
        "sasl.oauthbearer.clock.skew.seconds": "30",
        "zookeeper.connect": "192.168.50.178:18969",
        "zookeeper.ssl.ocsp.enable": "false",
        "broker.heartbeat.interval.ms": "2000",
        "authorizer.class.name": "",
        "sasl.mechanism.controller.protocol": "GSSAPI",
        "log.cleaner.max.compaction.lag.ms": "9223372036854775807",
        "num.replica.fetchers": "1",
        "alter.log.dirs.replication.quota.window.size.seconds": "1",
        "alter.log.dirs.replication.quota.window.num": "11",
        "log.roll.jitter.hours": "0",
        "log.cleaner.enable": "true",
        "controller.quorum.fetch.timeout.ms": "2000",
        "offsets.load.buffer.size": "5242880",
        "log.cleaner.delete.retention.ms": "86400000",
        "ssl.client.auth": "none",
        "controlled.shutdown.max.retries": "3",
        "sasl.login.retry.backoff.ms": "100",
        "queued.max.requests": "500",
        "offsets.topic.replication.factor": "1",
        "log.cleaner.threads": "1",
        "transaction.state.log.min.isr": "1",
        "sasl.kerberos.ticket.renew.jitter": "0.05",
        "socket.request.max.bytes": "104857600",
        "ssl.trustmanager.algorithm": "PKIX",
        "zookeeper.session.timeout.ms": "18000",
        "log.retention.bytes": "-1",
        "controller.quota.window.size.seconds": "1",
        "log.message.timestamp.type": "CreateTime",
        "sasl.kerberos.min.time.before.relogin": "60000",
        "connections.max.idle.ms": "600000",
        "zookeeper.set.acl": "false",
        "offsets.retention.minutes": "10080",
        "delegation.token.expiry.time.ms": "86400000",
        "max.connections": "2147483647",
        "transaction.state.log.num.partitions": "50",
        "replica.fetch.backoff.ms": "1000",
        "inter.broker.protocol.version": "3.1-IV0",
        "kafka.metrics.reporters": "",
        "controller.quorum.election.timeout.ms": "1000",
        "listener.security.protocol.map": "PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL",
        "log.retention.hours": "168",
        "num.partitions": "8",
        "socket.connection.setup.timeout.ms": "10000",
        "broker.id.generation.enable": "true",
        "listeners": "PLAINTEXT://:9092",
        "ssl.enabled.protocols": "TLSv1.2,TLSv1.3",
        "delete.records.purgatory.purge.interval.requests": "1",
        "delegation.token.expiry.check.interval.ms": "3600000",
        "ssl.cipher.suites": "",
        "log.flush.scheduler.interval.ms": "9223372036854775807",
        "zookeeper.max.in.flight.requests": "10",
        "controller.quorum.retry.backoff.ms": "20",
        "log.index.size.max.bytes": "10485760",
        "ssl.keymanager.algorithm": "SunX509",
        "broker.session.timeout.ms": "9000",
        "security.inter.broker.protocol": "PLAINTEXT",
        "replica.fetch.max.bytes": "1048576",
        "zookeeper.ssl.crl.enable": "false",
        "log.cleaner.dedupe.buffer.size": "134217728",
        "node.id": "-1",
        "replica.high.watermark.checkpoint.interval.ms": "5000",
        "replication.quota.window.size.seconds": "1",
        "log.cleaner.io.buffer.size": "524288",
        "sasl.kerberos.ticket.renew.window.factor": "0.8",
        "metrics.recording.level": "INFO",
        "password.encoder.cipher.algorithm": "AES/CBC/PKCS5Padding",
        "controlled.shutdown.retry.backoff.ms": "5000",
        "log.roll.hours": "168",
        "log.cleanup.policy": "delete",
        "initial.broker.registration.timeout.ms": "60000",
        "log.flush.start.offset.checkpoint.interval.ms": "60000",
        "ssl.principal.mapping.rules": "DEFAULT",
        "transaction.state.log.segment.bytes": "104857600",
        "max.connections.per.ip": "2147483647",
        "offsets.topic.segment.bytes": "104857600",
        "background.threads": "10",
        "request.timeout.ms": "30000",
        "sasl.login.retry.backoff.max.ms": "10000",
        "log.message.format.version": "3.0-IV1",
        "group.initial.rebalance.delay.ms": "3000",
        "log.index.interval.bytes": "4096",
        "log.dir": "/tmp/kafka-logs",
        "log.segment.bytes": "1073741824",
        "log.cleaner.backoff.ms": "15000",
        "offset.metadata.max.bytes": "4096",
        "replica.fetch.response.max.bytes": "10485760",
        "group.max.session.timeout.ms": "1800000",
        "zookeeper.sync.time.ms": "2000",
        "controller.quorum.append.linger.ms": "25",
        "log.segment.delete.delay.ms": "60000",
        "fetch.max.bytes": "57671680",
        "metadata.max.retention.bytes": "-1",
        "log.dirs": "/tmp/log-folder-0,/tmp/log-folder-1,/tmp/log-folder-2",
        "controlled.shutdown.enable": "true",
        "compression.type": "producer",
        "max.connections.per.ip.overrides": "",
        "socket.connection.setup.timeout.max.ms": "30000",
        "log.message.timestamp.difference.max.ms": "9223372036854775807",
        "sasl.oauthbearer.scope.claim.name": "scope",
        "password.encoder.key.length": "128",
        "sasl.login.refresh.min.period.seconds": "60",
        "sasl.login.refresh.window.factor": "0.8",
        "kafka.metrics.polling.interval.secs": "10",
        "metadata.log.max.record.bytes.between.snapshots": "20971520",
        "transaction.abort.timed.out.transaction.cleanup.interval.ms": "10000",
        "sasl.kerberos.kinit.cmd": "/usr/bin/kinit",
        "metadata.max.retention.ms": "604800000",
        "log.cleaner.io.max.bytes.per.second": "1.7976931348623157E308",
        "auto.leader.rebalance.enable": "true",
        "leader.imbalance.check.interval.seconds": "300",
        "controller.quorum.election.backoff.max.ms": "1000",
        "log.cleaner.min.cleanable.ratio": "0.5",
        "replica.lag.time.max.ms": "30000",
        "max.incremental.fetch.session.cache.slots": "1000",
        "num.network.threads": "8",
        "reserved.broker.max.id": "1000",
        "metrics.num.samples": "2",
        "transaction.remove.expired.transaction.cleanup.interval.ms": "3600000",
        "socket.send.buffer.bytes": "102400",
        "log.message.downconversion.enable": "true",
        "ssl.protocol": "TLSv1.3",
        "metadata.log.segment.ms": "604800000",
        "transaction.state.log.load.buffer.size": "5242880",
        "socket.receive.buffer.bytes": "102400",
        "sasl.oauthbearer.sub.claim.name": "sub",
        "replica.fetch.min.bytes": "1",
        "unclean.leader.election.enable": "false",
        "sasl.enabled.mechanisms": "GSSAPI",
        "sasl.oauthbearer.jwks.endpoint.retry.backoff.ms": "100",
        "group.min.session.timeout.ms": "6000",
        "offsets.retention.check.interval.ms": "600000",
        "log.cleaner.io.buffer.load.factor": "0.9",
        "transaction.max.timeout.ms": "900000",
        "producer.purgatory.purge.interval.requests": "1000",
        "controller.quorum.voters": "",
        "metrics.sample.window.ms": "30000",
        "process.roles": "",
        "group.max.size": "2147483647",
        "sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms": "10000",
        "delegation.token.max.lifetime.ms": "604800000",
        "broker.id": "-1",
        "offsets.topic.compression.codec": "0",
        "zookeeper.ssl.endpoint.identification.algorithm": "HTTPS",
        "replication.quota.window.num": "11",
        "log.retention.check.interval.ms": "300000",
        "advertised.listeners": "PLAINTEXT://192.168.50.178:10261",
        "sasl.login.refresh.window.jitter": "0.05",
        "leader.imbalance.per.broker.percentage": "10",
        "queued.max.request.bytes": "-1",
        "controller.quota.window.num": "11"
      }
    }
  ]
}
```


Patch:
diff --git a/app/src/main/java/org/astraea/web/BrokerHandler.java b/app/src/main/java/org/astraea/web/BrokerHandler.java
new file mode 100644
index 00000000..d723b7fb
--- /dev/null
+++ b/app/src/main/java/org/astraea/web/BrokerHandler.java
@@ -0,0 +1,65 @@
+package org.astraea.web;
+
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+import org.astraea.admin.Admin;
+import org.astraea.admin.Config;
+
+class BrokerHandler implements Handler {
+
+  private final Admin admin;
+
+  BrokerHandler(Admin admin) {
+    this.admin = admin;
+  }
+
+  private static void mustBeNumber(Optional<String> target) {
+    try {
+      target.ifPresent(Integer::valueOf);
+    } catch (NumberFormatException e) {
+      throw new NoSuchElementException("broker: " + target.get() + " does not exist");
+    }
+  }
+
+  @Override
+  public JsonObject response(Optional<String> target, Map<String, String> queries) {
+    mustBeNumber(target);
+    Predicate<Map.Entry<Integer, ?>> brokerId =
+        e -> target.stream().map(Integer::valueOf).allMatch(t -> t.equals(e.getKey()));
+    var brokers =
+        admin.brokers().entrySet().stream()
+            .filter(brokerId)
+            .map(e -> new Broker(e.getKey(), e.getValue()))
+            .collect(Collectors.toUnmodifiableList());
+
+    if (target.isPresent() && brokers.size() == 1) return brokers.get(0);
+    else if (target.isPresent())
+      throw new NoSuchElementException("broker: " + target.get() + " does not exist");
+    else return new Brokers(brokers);
+  }
+
+  static class Broker implements JsonObject {
+    final int id;
+    final Map<String, String> configs;
+
+    Broker(int id, Config configs) {
+      this.id = id;
+      this.configs =
+          StreamSupport.stream(configs.spliterator(), false)
+              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    }
+  }
+
+  static class Brokers implements JsonObject {
+    final List<Broker> brokers;
+
+    Brokers(List<Broker> brokers) {
+      this.brokers = brokers;
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/web/WebService.java b/app/src/main/java/org/astraea/web/WebService.java
index 944309a9..80fe8b53 100644
--- a/app/src/main/java/org/astraea/web/WebService.java
+++ b/app/src/main/java/org/astraea/web/WebService.java
@@ -15,8 +15,9 @@ public class WebService {
 
   private static void execute(Argument arg) throws IOException {
     var server = HttpServer.create(new InetSocketAddress(arg.port), 0);
-    server.createContext("/topics", new TopicHandler(Admin.of(arg.bootstrapServers())));
-    server.createContext("/groups", new GroupHandler(Admin.of(arg.bootstrapServers())));
+    server.createContext("/topics", new TopicHandler(Admin.of(arg.configs())));
+    server.createContext("/groups", new GroupHandler(Admin.of(arg.configs())));
+    server.createContext("/brokers", new BrokerHandler(Admin.of(arg.configs())));
     server.start();
   }
 


Test patch:
diff --git a/app/src/test/java/org/astraea/service/RequireBrokerCluster.java b/app/src/test/java/org/astraea/service/RequireBrokerCluster.java
index 40c5866f..b278d59f 100644
--- a/app/src/test/java/org/astraea/service/RequireBrokerCluster.java
+++ b/app/src/test/java/org/astraea/service/RequireBrokerCluster.java
@@ -21,6 +21,10 @@ public abstract class RequireBrokerCluster extends RequireJmxServer {
     return BROKER_CLUSTER.logFolders();
   }
 
+  protected static Set<Integer> brokerIds() {
+    return logFolders().keySet();
+  }
+
   @AfterAll
   static void shutdownClusters() {
     Utils.close(BROKER_CLUSTER);
diff --git a/app/src/test/java/org/astraea/web/BrokerHandlerTest.java b/app/src/test/java/org/astraea/web/BrokerHandlerTest.java
new file mode 100644
index 00000000..3275ed63
--- /dev/null
+++ b/app/src/test/java/org/astraea/web/BrokerHandlerTest.java
@@ -0,0 +1,64 @@
+package org.astraea.web;
+
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import org.astraea.admin.Admin;
+import org.astraea.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class BrokerHandlerTest extends RequireBrokerCluster {
+
+  @Test
+  void testListBrokers() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new BrokerHandler(admin);
+      var response =
+          Assertions.assertInstanceOf(
+              BrokerHandler.Brokers.class, handler.response(Optional.empty(), Map.of()));
+      Assertions.assertEquals(brokerIds().size(), response.brokers.size());
+      brokerIds()
+          .forEach(
+              id ->
+                  Assertions.assertEquals(
+                      1, response.brokers.stream().filter(b -> b.id == id).count()));
+    }
+  }
+
+  @Test
+  void testQueryNonexistentBroker() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new BrokerHandler(admin);
+      var exception =
+          Assertions.assertThrows(
+              NoSuchElementException.class, () -> handler.response(Optional.of("99999"), Map.of()));
+      Assertions.assertTrue(exception.getMessage().contains("99999"));
+    }
+  }
+
+  @Test
+  void testQueryInvalidBroker() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new BrokerHandler(admin);
+      var exception =
+          Assertions.assertThrows(
+              NoSuchElementException.class, () -> handler.response(Optional.of("abc"), Map.of()));
+      Assertions.assertTrue(exception.getMessage().contains("abc"));
+    }
+  }
+
+  @Test
+  void testQuerySingleBroker() throws InterruptedException {
+    var brokerId = brokerIds().iterator().next();
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new BrokerHandler(admin);
+      var broker =
+          Assertions.assertInstanceOf(
+              BrokerHandler.Broker.class,
+              handler.response(Optional.of(String.valueOf(brokerId)), Map.of()));
+      Assertions.assertEquals(brokerId, broker.id);
+      Assertions.assertNotEquals(0, broker.configs.size());
+    }
+  }
+}