Repository: skiptests/astraea
Pull request #334 (2022-05-13 09:28:52)

Enable to get broker configuration
這讓我們的admin可以列舉broker參數，下一隻PR會把這個功能安裝到web service上面


Patch:
diff --git a/app/src/main/java/org/astraea/admin/TopicAdmin.java b/app/src/main/java/org/astraea/admin/Admin.java
similarity index 79%
rename from app/src/main/java/org/astraea/admin/TopicAdmin.java
rename to app/src/main/java/org/astraea/admin/Admin.java
index 93902bc8..76fbe0ff 100644
--- a/app/src/main/java/org/astraea/admin/TopicAdmin.java
+++ b/app/src/main/java/org/astraea/admin/Admin.java
@@ -7,17 +7,17 @@ import java.util.Map;
 import java.util.Set;
 import org.apache.kafka.common.TopicPartition;
 
-public interface TopicAdmin extends Closeable {
+public interface Admin extends Closeable {
 
   static Builder builder() {
     return new Builder();
   }
 
-  static TopicAdmin of(String bootstrapServers) {
+  static Admin of(String bootstrapServers) {
     return builder().bootstrapServers(bootstrapServers).build();
   }
 
-  static TopicAdmin of(Map<String, String> configs) {
+  static Admin of(Map<String, String> configs) {
     return builder().configs(configs).build();
   }
 
@@ -26,14 +26,14 @@ public interface TopicAdmin extends Closeable {
   }
 
   /** @return the topic name and its configurations. */
-  Map<String, TopicConfig> topics();
+  Map<String, Config> topics();
 
   default Set<String> publicTopicNames() {
     return publicTopics().keySet();
   }
 
   /** @return the topic name and its configurations. (Without internal topics) */
-  Map<String, TopicConfig> publicTopics();
+  Map<String, Config> publicTopics();
 
   /** @return a topic creator to set all topic configs and then run the procedure. */
   Creator creator();
@@ -45,16 +45,15 @@ public interface TopicAdmin extends Closeable {
   Map<TopicPartition, Offset> offsets(Set<String> topics);
 
   /** @return all consumer groups */
-  default Map<String, ConsumerGroup> consumerGroup() {
-    return consumerGroup(Set.of());
+  default Map<String, ConsumerGroup> consumerGroups() {
+    return consumerGroups(Set.of());
   }
 
   /**
-   * @param consumerGroupNames consumer group names. if empty set given, every consume group will
-   *     return.
+   * @param consumerGroupNames consumer group names.
    * @return the member info of each consumer group
    */
-  Map<String, ConsumerGroup> consumerGroup(Set<String> consumerGroupNames);
+  Map<String, ConsumerGroup> consumerGroups(Set<String> consumerGroupNames);
 
   /**
    * @param topics topic names
@@ -62,6 +61,9 @@ public interface TopicAdmin extends Closeable {
    */
   Map<TopicPartition, List<Replica>> replicas(Set<String> topics);
 
+  /** @return all broker id and their configuration */
+  Map<Integer, Config> brokers();
+
   /** @return all brokers' ids */
   Set<Integer> brokerIds();
 
diff --git a/app/src/main/java/org/astraea/admin/Builder.java b/app/src/main/java/org/astraea/admin/Builder.java
index babcbb58..13f388f4 100644
--- a/app/src/main/java/org/astraea/admin/Builder.java
+++ b/app/src/main/java/org/astraea/admin/Builder.java
@@ -18,7 +18,6 @@ import java.util.concurrent.TimeoutException;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.stream.Collectors;
-import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.ConsumerGroupListing;
@@ -53,14 +52,14 @@ public class Builder {
     return this;
   }
 
-  public TopicAdmin build() {
-    return new TopicAdminImpl(Admin.create(configs));
+  public Admin build() {
+    return new AdminImpl(org.apache.kafka.clients.admin.Admin.create(configs));
   }
 
-  private static class TopicAdminImpl implements TopicAdmin {
-    private final Admin admin;
+  private static class AdminImpl implements Admin {
+    private final org.apache.kafka.clients.admin.Admin admin;
 
-    TopicAdminImpl(Admin admin) {
+    AdminImpl(org.apache.kafka.clients.admin.Admin admin) {
       this.admin = Objects.requireNonNull(admin);
     }
 
@@ -92,7 +91,7 @@ public class Builder {
     }
 
     @Override
-    public Map<String, ConsumerGroup> consumerGroup(Set<String> consumerGroupNames) {
+    public Map<String, ConsumerGroup> consumerGroups(Set<String> consumerGroupNames) {
 
       final Set<String> groupIds =
           !consumerGroupNames.isEmpty()
@@ -177,16 +176,16 @@ public class Builder {
     }
 
     @Override
-    public Map<String, TopicConfig> topics() {
+    public Map<String, Config> topics() {
       return topics(true);
     }
 
     @Override
-    public Map<String, TopicConfig> publicTopics() {
+    public Map<String, Config> publicTopics() {
       return topics(false);
     }
 
-    private Map<String, TopicConfig> topics(boolean listInternal) {
+    private Map<String, Config> topics(boolean listInternal) {
       var topics =
           Utils.handleException(
               () ->
@@ -205,13 +204,29 @@ public class Builder {
                       .get())
           .entrySet()
           .stream()
+          .collect(Collectors.toMap(e -> e.getKey().name(), e -> new ConfigImpl(e.getValue())));
+    }
+
+    @Override
+    public Map<Integer, Config> brokers() {
+      var brokerIds = brokerIds();
+      return Utils.handleException(
+              () ->
+                  admin
+                      .describeConfigs(
+                          brokerIds.stream()
+                              .map(
+                                  id ->
+                                      new ConfigResource(
+                                          ConfigResource.Type.BROKER, String.valueOf(id)))
+                              .collect(Collectors.toList()))
+                      .all()
+                      .get())
+          .entrySet()
+          .stream()
           .collect(
               Collectors.toMap(
-                  e -> e.getKey().name(),
-                  e ->
-                      new TopicConfigImpl(
-                          e.getValue().entries().stream()
-                              .collect(Collectors.toMap(ConfigEntry::name, ConfigEntry::value)))));
+                  e -> Integer.valueOf(e.getKey().name()), e -> new ConfigImpl(e.getValue())));
     }
 
     @Override
@@ -335,10 +350,17 @@ public class Builder {
     }
   }
 
-  private static class TopicConfigImpl implements TopicConfig {
+  private static class ConfigImpl implements Config {
     private final Map<String, String> configs;
 
-    TopicConfigImpl(Map<String, String> configs) {
+    ConfigImpl(org.apache.kafka.clients.admin.Config config) {
+      this(
+          config.entries().stream()
+              .filter(e -> e.value() != null)
+              .collect(Collectors.toMap(ConfigEntry::name, ConfigEntry::value)));
+    }
+
+    ConfigImpl(Map<String, String> configs) {
       this.configs = Collections.unmodifiableMap(configs);
     }
 
@@ -364,18 +386,18 @@ public class Builder {
   }
 
   private static class CreatorImpl implements Creator {
-    private final Admin admin;
+    private final org.apache.kafka.clients.admin.Admin admin;
     private final Function<String, Map<TopicPartition, List<Replica>>> replicasGetter;
-    private final Function<String, TopicConfig> configsGetter;
+    private final Function<String, Config> configsGetter;
     private String topic;
     private int numberOfPartitions = 1;
     private short numberOfReplicas = 1;
     private final Map<String, String> configs = new HashMap<>();
 
     CreatorImpl(
-        Admin admin,
+        org.apache.kafka.clients.admin.Admin admin,
         Function<String, Map<TopicPartition, List<Replica>>> replicasGetter,
-        Function<String, TopicConfig> configsGetter) {
+        Function<String, Config> configsGetter) {
       this.admin = admin;
       this.replicasGetter = replicasGetter;
       this.configsGetter = configsGetter;
@@ -468,11 +490,13 @@ public class Builder {
   }
 
   private static class MigratorImpl implements Migrator {
-    private final Admin admin;
+    private final org.apache.kafka.clients.admin.Admin admin;
     private final Function<Set<String>, Set<TopicPartition>> partitionGetter;
     private final Set<TopicPartition> partitions = new HashSet<>();
 
-    MigratorImpl(Admin admin, Function<Set<String>, Set<TopicPartition>> partitionGetter) {
+    MigratorImpl(
+        org.apache.kafka.clients.admin.Admin admin,
+        Function<Set<String>, Set<TopicPartition>> partitionGetter) {
       this.admin = admin;
       this.partitionGetter = partitionGetter;
     }
diff --git a/app/src/main/java/org/astraea/admin/Config.java b/app/src/main/java/org/astraea/admin/Config.java
new file mode 100644
index 00000000..0011d661
--- /dev/null
+++ b/app/src/main/java/org/astraea/admin/Config.java
@@ -0,0 +1,21 @@
+package org.astraea.admin;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/** this interface used to represent the resource (topic or broker) configuration. */
+public interface Config extends Iterable<Map.Entry<String, String>> {
+  /**
+   * @param key config key
+   * @return the value associated to input key. otherwise, empty
+   */
+  Optional<String> value(String key);
+
+  /** @return all keys in this configuration */
+  Set<String> keys();
+
+  /** @return all values in this configuration */
+  Collection<String> values();
+}
diff --git a/app/src/main/java/org/astraea/admin/ReplicaCollie.java b/app/src/main/java/org/astraea/admin/ReplicaCollie.java
index caf5e039..e063694b 100644
--- a/app/src/main/java/org/astraea/admin/ReplicaCollie.java
+++ b/app/src/main/java/org/astraea/admin/ReplicaCollie.java
@@ -26,7 +26,7 @@ public class ReplicaCollie {
     Set<String> pathSink;
   }
 
-  static Argument setArguments(TopicAdmin admin, Argument args) {
+  static Argument setArguments(Admin admin, Argument args) {
     Argument argument = new Argument();
     argument.topics = args.topics.isEmpty() ? admin.topicNames() : args.topics;
     argument.fromBrokers = args.fromBrokers;
@@ -48,7 +48,7 @@ public class ReplicaCollie {
     return argument;
   }
 
-  static void checkArgs(TopicAdmin admin, Argument args) {
+  static void checkArgs(Admin admin, Argument args) {
     var topics = args.topics.isEmpty() ? admin.topicNames() : args.topics;
     var path = args.path.isEmpty() ? null : args.path.iterator().next();
     var targetBrokers = args.toBrokers.isEmpty() ? args.fromBrokers : args.toBrokers;
@@ -89,7 +89,7 @@ public class ReplicaCollie {
   }
 
   static TreeMap<TopicPartition, Map.Entry<List<Integer>, List<Integer>>> checkMigratorBroker(
-      TopicAdmin admin, Argument argument) {
+      Admin admin, Argument argument) {
     TreeMap<TopicPartition, Map.Entry<List<Integer>, List<Integer>>> brokerMigrate =
         new TreeMap<>(
             Comparator.comparing(TopicPartition::topic).thenComparing(TopicPartition::partition));
@@ -139,7 +139,7 @@ public class ReplicaCollie {
   }
 
   static TreeMap<TopicPartition, Map.Entry<Set<String>, Set<String>>> checkMigratorPath(
-      TopicAdmin admin, Argument argument) {
+      Admin admin, Argument argument) {
     TreeMap<TopicPartition, Map.Entry<Set<String>, Set<String>>> pathMigrate =
         new TreeMap<>(
             Comparator.comparing(TopicPartition::topic).thenComparing(TopicPartition::partition));
@@ -185,8 +185,7 @@ public class ReplicaCollie {
   }
 
   static void brokerMigrator(
-      TreeMap<TopicPartition, Map.Entry<List<Integer>, List<Integer>>> brokerMigrate,
-      TopicAdmin admin) {
+      TreeMap<TopicPartition, Map.Entry<List<Integer>, List<Integer>>> brokerMigrate, Admin admin) {
     brokerMigrate.forEach(
         (tp, assignments) -> {
           if (!assignments.getKey().equals(assignments.getValue())) {
@@ -197,7 +196,7 @@ public class ReplicaCollie {
 
   static void pathMigrator(
       TreeMap<TopicPartition, Map.Entry<Set<String>, Set<String>>> pathMigrate,
-      TopicAdmin admin,
+      Admin admin,
       Integer broker) {
     pathMigrate.forEach(
         (tp, assignments) -> {
@@ -213,7 +212,7 @@ public class ReplicaCollie {
       TreeMap<TopicPartition, Map.Entry<List<Integer>, List<Integer>>> brokerMigrate,
       TreeMap<TopicPartition, Map.Entry<Set<String>, Set<String>>> pathMigrate,
       Argument argument,
-      TopicAdmin admin) {
+      Admin admin) {
     var result =
         new TreeMap<TopicPartition, MigratorInfo>(
             Comparator.comparing(TopicPartition::topic).thenComparing(TopicPartition::partition));
@@ -283,7 +282,7 @@ public class ReplicaCollie {
     return result;
   }
 
-  static Map<TopicPartition, MigratorInfo> execute(TopicAdmin admin, Argument args) {
+  static Map<TopicPartition, MigratorInfo> execute(Admin admin, Argument args) {
     checkArgs(admin, args);
     Argument argument = setArguments(admin, args);
     var brokerMigrate = checkMigratorBroker(admin, argument);
@@ -298,7 +297,7 @@ public class ReplicaCollie {
   public static void main(String[] args) throws IOException {
     var argument = org.astraea.argument.Argument.parse(new Argument(), args);
 
-    try (var admin = TopicAdmin.of(argument.bootstrapServers())) {
+    try (var admin = Admin.of(argument.bootstrapServers())) {
       execute(admin, argument)
           .forEach(
               (tp, assignments) ->
diff --git a/app/src/main/java/org/astraea/admin/ReplicaSyncingMonitor.java b/app/src/main/java/org/astraea/admin/ReplicaSyncingMonitor.java
index 4cde57b6..a0329431 100644
--- a/app/src/main/java/org/astraea/admin/ReplicaSyncingMonitor.java
+++ b/app/src/main/java/org/astraea/admin/ReplicaSyncingMonitor.java
@@ -28,12 +28,12 @@ public class ReplicaSyncingMonitor {
 
   public static void main(String[] args) {
     Argument argument = org.astraea.argument.Argument.parse(new Argument(), args);
-    try (TopicAdmin topicAdmin = TopicAdmin.of(argument.bootstrapServers())) {
+    try (Admin topicAdmin = Admin.of(argument.bootstrapServers())) {
       execute(topicAdmin, argument);
     }
   }
 
-  static void execute(final TopicAdmin topicAdmin, final Argument argument) {
+  static void execute(final Admin topicAdmin, final Argument argument) {
 
     // this supplier will give you all the topic names that the client is interested in.
     // discover any newly happened non-synced replica can be a quiet useful scenario, so here we use
@@ -193,7 +193,7 @@ public class ReplicaSyncingMonitor {
   }
 
   static Set<TopicPartition> findNonSyncedTopicPartition(
-      TopicAdmin topicAdmin, Set<String> topicToTrack) {
+      Admin topicAdmin, Set<String> topicToTrack) {
     return topicAdmin.replicas(topicToTrack).entrySet().stream()
         .filter(tpr -> tpr.getValue().stream().anyMatch(replica -> !replica.inSync()))
         .map(Map.Entry::getKey)
diff --git a/app/src/main/java/org/astraea/admin/TopicConfig.java b/app/src/main/java/org/astraea/admin/TopicConfig.java
deleted file mode 100644
index 8cd4548a..00000000
--- a/app/src/main/java/org/astraea/admin/TopicConfig.java
+++ /dev/null
@@ -1,15 +0,0 @@
-package org.astraea.admin;
-
-import java.util.Collection;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-
-public interface TopicConfig extends Iterable<Map.Entry<String, String>> {
-
-  Optional<String> value(String key);
-
-  Set<String> keys();
-
-  Collection<String> values();
-}
diff --git a/app/src/main/java/org/astraea/admin/TopicExplorer.java b/app/src/main/java/org/astraea/admin/TopicExplorer.java
index e95f31ba..f8847a5b 100644
--- a/app/src/main/java/org/astraea/admin/TopicExplorer.java
+++ b/app/src/main/java/org/astraea/admin/TopicExplorer.java
@@ -61,7 +61,7 @@ public class TopicExplorer {
     }
   }
 
-  static Result execute(TopicAdmin admin, Set<String> topics) {
+  static Result execute(Admin admin, Set<String> topics) {
     var replicas = admin.replicas(topics);
 
     var invalidTopics =
@@ -79,7 +79,7 @@ public class TopicExplorer {
     }
 
     var offsets = admin.offsets(topics);
-    var consumerGroups = admin.consumerGroup(Set.of());
+    var consumerGroups = admin.consumerGroups(Set.of());
     var time = LocalDateTime.now();
 
     // Given topic name, return the partition count
@@ -113,7 +113,7 @@ public class TopicExplorer {
 
   public static void main(String[] args) throws IOException {
     var argument = org.astraea.argument.Argument.parse(new Argument(), args);
-    try (var admin = TopicAdmin.of(argument.bootstrapServers())) {
+    try (var admin = Admin.of(argument.bootstrapServers())) {
       var result = execute(admin, argument.topics.isEmpty() ? admin.topicNames() : argument.topics);
       TreeOutput.print(result, System.out);
     }
diff --git a/app/src/main/java/org/astraea/cost/topic/GetPartitionInf.java b/app/src/main/java/org/astraea/cost/topic/GetPartitionInf.java
index 5e6a6496..65c643d4 100644
--- a/app/src/main/java/org/astraea/cost/topic/GetPartitionInf.java
+++ b/app/src/main/java/org/astraea/cost/topic/GetPartitionInf.java
@@ -7,10 +7,10 @@ import java.util.NoSuchElementException;
 import java.util.TreeMap;
 import java.util.stream.Collectors;
 import org.apache.kafka.common.TopicPartition;
-import org.astraea.admin.TopicAdmin;
+import org.astraea.admin.Admin;
 
 public class GetPartitionInf {
-  static Map<Integer, Map<TopicPartition, Integer>> getSize(TopicAdmin client) {
+  static Map<Integer, Map<TopicPartition, Integer>> getSize(Admin client) {
     Map<Integer, Map<TopicPartition, Integer>> brokerPartitionSize = new HashMap<>();
     client
         .brokerIds()
@@ -35,7 +35,7 @@ public class GetPartitionInf {
     return brokerPartitionSize;
   }
 
-  static Map<String, Integer> getRetentionMillis(TopicAdmin client) {
+  static Map<String, Integer> getRetentionMillis(Admin client) {
     return client.topics().entrySet().stream()
         .collect(
             Collectors.toMap(
diff --git a/app/src/main/java/org/astraea/cost/topic/PartitionScore.java b/app/src/main/java/org/astraea/cost/topic/PartitionScore.java
index 0fb29b04..b9618248 100644
--- a/app/src/main/java/org/astraea/cost/topic/PartitionScore.java
+++ b/app/src/main/java/org/astraea/cost/topic/PartitionScore.java
@@ -9,7 +9,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.kafka.common.TopicPartition;
-import org.astraea.admin.TopicAdmin;
+import org.astraea.admin.Admin;
 import org.astraea.argument.BooleanField;
 
 public class PartitionScore {
@@ -63,8 +63,7 @@ public class PartitionScore {
     }
   }
 
-  public static Map<Integer, Map<TopicPartition, Double>> execute(
-      Argument argument, TopicAdmin admin) {
+  public static Map<Integer, Map<TopicPartition, Double>> execute(Argument argument, Admin admin) {
     var internalTopic =
         Set.of(
             "__consumer_offsets",
@@ -81,7 +80,7 @@ public class PartitionScore {
 
   public static void main(String[] args) {
     var argument = org.astraea.argument.Argument.parse(new Argument(), args);
-    var admin = TopicAdmin.of(argument.bootstrapServers());
+    var admin = Admin.of(argument.bootstrapServers());
     var score = execute(argument, admin);
     printScore(score, argument);
   }
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index 0d9b999a..bb552ea7 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -18,8 +18,8 @@ import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.WakeupException;
 import org.astraea.Utils;
+import org.astraea.admin.Admin;
 import org.astraea.admin.Compression;
-import org.astraea.admin.TopicAdmin;
 import org.astraea.argument.CompressionField;
 import org.astraea.argument.NonEmptyStringField;
 import org.astraea.argument.NonNegativeShortField;
@@ -112,7 +112,7 @@ public class Performance {
   public static Result execute(final Argument param)
       throws InterruptedException, IOException, ExecutionException {
     List<Integer> partitions;
-    try (var topicAdmin = TopicAdmin.of(param.configs())) {
+    try (var topicAdmin = Admin.of(param.configs())) {
       topicAdmin
           .creator()
           .numberOfReplicas(param.replicas)
@@ -232,7 +232,7 @@ public class Performance {
   }
 
   // visible for test
-  static List<Integer> partition(Argument param, TopicAdmin topicAdmin) {
+  static List<Integer> partition(Argument param, Admin topicAdmin) {
     if (positiveSpecifyBroker(param)) {
       return topicAdmin
           .partitionsOfBrokers(Set.of(param.topic), new HashSet<>(param.specifyBroker))
diff --git a/app/src/main/java/org/astraea/web/GroupHandler.java b/app/src/main/java/org/astraea/web/GroupHandler.java
index a5d3bf45..fe9ceee4 100644
--- a/app/src/main/java/org/astraea/web/GroupHandler.java
+++ b/app/src/main/java/org/astraea/web/GroupHandler.java
@@ -6,13 +6,13 @@ import java.util.NoSuchElementException;
 import java.util.Optional;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
-import org.astraea.admin.TopicAdmin;
+import org.astraea.admin.Admin;
 
 public class GroupHandler implements Handler {
 
-  private final TopicAdmin admin;
+  private final Admin admin;
 
-  GroupHandler(TopicAdmin admin) {
+  GroupHandler(Admin admin) {
     this.admin = admin;
   }
 
@@ -21,7 +21,7 @@ public class GroupHandler implements Handler {
     Predicate<Map.Entry<String, ?>> groupFilter =
         e -> target.stream().allMatch(t -> t.equals(e.getKey()));
     var topics = admin.topicNames();
-    var consumerGroups = admin.consumerGroup();
+    var consumerGroups = admin.consumerGroups();
     var offsets = admin.offsets(topics);
 
     var groups =
diff --git a/app/src/main/java/org/astraea/web/TopicHandler.java b/app/src/main/java/org/astraea/web/TopicHandler.java
index 6591e678..01da67a6 100644
--- a/app/src/main/java/org/astraea/web/TopicHandler.java
+++ b/app/src/main/java/org/astraea/web/TopicHandler.java
@@ -8,14 +8,14 @@ import java.util.Optional;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
-import org.astraea.admin.TopicAdmin;
-import org.astraea.admin.TopicConfig;
+import org.astraea.admin.Admin;
+import org.astraea.admin.Config;
 
 class TopicHandler implements Handler {
 
-  private final TopicAdmin admin;
+  private final Admin admin;
 
-  TopicHandler(TopicAdmin admin) {
+  TopicHandler(Admin admin) {
     this.admin = admin;
   }
 
@@ -66,7 +66,7 @@ class TopicHandler implements Handler {
     final List<Partition> partitions;
     final Map<String, String> configs;
 
-    private TopicInfo(String name, List<Partition> partitions, TopicConfig configs) {
+    private TopicInfo(String name, List<Partition> partitions, Config configs) {
       this.name = name;
       this.partitions = partitions;
       this.configs =
diff --git a/app/src/main/java/org/astraea/web/WebService.java b/app/src/main/java/org/astraea/web/WebService.java
index 72d1ee73..944309a9 100644
--- a/app/src/main/java/org/astraea/web/WebService.java
+++ b/app/src/main/java/org/astraea/web/WebService.java
@@ -4,7 +4,7 @@ import com.beust.jcommander.Parameter;
 import com.sun.net.httpserver.HttpServer;
 import java.io.IOException;
 import java.net.InetSocketAddress;
-import org.astraea.admin.TopicAdmin;
+import org.astraea.admin.Admin;
 import org.astraea.argument.NonNegativeIntegerField;
 
 public class WebService {
@@ -15,8 +15,8 @@ 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(TopicAdmin.of(arg.bootstrapServers())));
-    server.createContext("/groups", new GroupHandler(TopicAdmin.of(arg.bootstrapServers())));
+    server.createContext("/topics", new TopicHandler(Admin.of(arg.bootstrapServers())));
+    server.createContext("/groups", new GroupHandler(Admin.of(arg.bootstrapServers())));
     server.start();
   }
 


Test patch:
diff --git a/app/src/test/java/org/astraea/admin/TopicAdminTest.java b/app/src/test/java/org/astraea/admin/AdminTest.java
similarity index 92%
rename from app/src/test/java/org/astraea/admin/TopicAdminTest.java
rename to app/src/test/java/org/astraea/admin/AdminTest.java
index 841a69d9..12a89b82 100644
--- a/app/src/test/java/org/astraea/admin/TopicAdminTest.java
+++ b/app/src/test/java/org/astraea/admin/AdminTest.java
@@ -24,12 +24,12 @@ import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.condition.DisabledOnOs;
 
-public class TopicAdminTest extends RequireBrokerCluster {
+public class AdminTest extends RequireBrokerCluster {
 
   @Test
   void testCreator() {
     var topicName = "testCreator";
-    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+    try (var topicAdmin = Admin.of(bootstrapServers())) {
       topicAdmin
           .creator()
           .topic(topicName)
@@ -55,7 +55,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
   @Test
   void testCreateTopicRepeatedly() {
     var topicName = "testCreateTopicRepeatedly";
-    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+    try (var topicAdmin = Admin.of(bootstrapServers())) {
       Runnable createTopic =
           () ->
               topicAdmin
@@ -95,7 +95,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
   @Test
   void testPartitions() throws InterruptedException {
     var topicName = "testPartitions";
-    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+    try (var topicAdmin = Admin.of(bootstrapServers())) {
       topicAdmin.creator().topic(topicName).numberOfPartitions(3).create();
       // wait for syncing topic creation
       TimeUnit.SECONDS.sleep(5);
@@ -118,7 +118,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
   @Test
   void testOffsets() throws InterruptedException {
     var topicName = "testOffsets";
-    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+    try (var topicAdmin = Admin.of(bootstrapServers())) {
       topicAdmin.creator().topic(topicName).numberOfPartitions(3).create();
       // wait for syncing topic creation
       TimeUnit.SECONDS.sleep(5);
@@ -138,7 +138,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
   void testConsumerGroups() throws InterruptedException {
     var topicName = "testConsumerGroups-Topic";
     var consumerGroup = "testConsumerGroups-Group";
-    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+    try (var topicAdmin = Admin.of(bootstrapServers())) {
       topicAdmin.creator().topic(topicName).numberOfPartitions(3).create();
       Consumer.builder()
           .bootstrapServers(bootstrapServers())
@@ -147,7 +147,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
           .build();
       // wait for syncing topic creation
       TimeUnit.SECONDS.sleep(5);
-      var consumerGroupMap = topicAdmin.consumerGroup(Set.of(consumerGroup));
+      var consumerGroupMap = topicAdmin.consumerGroups(Set.of(consumerGroup));
       Assertions.assertEquals(1, consumerGroupMap.size());
       Assertions.assertTrue(consumerGroupMap.containsKey(consumerGroup));
       Assertions.assertEquals(consumerGroup, consumerGroupMap.get(consumerGroup).groupId());
@@ -160,7 +160,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
   @DisabledOnOs(WINDOWS)
   void testMigrateSinglePartition() throws InterruptedException {
     var topicName = "testMigrateSinglePartition";
-    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+    try (var topicAdmin = Admin.of(bootstrapServers())) {
       topicAdmin.creator().topic(topicName).numberOfPartitions(1).create();
       // wait for syncing topic creation
       TimeUnit.SECONDS.sleep(5);
@@ -213,7 +213,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
   void testChangeReplicaLeader() throws InterruptedException {
     var topicName = "testChangeReplicaLeader";
     var tp = new TopicPartition(topicName, 0);
-    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+    try (var topicAdmin = Admin.of(bootstrapServers())) {
       topicAdmin
           .creator()
           .topic(topicName)
@@ -244,7 +244,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
   @DisabledOnOs(WINDOWS)
   void testMigrateAllPartitions() throws InterruptedException {
     var topicName = "testMigrateAllPartitions";
-    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+    try (var topicAdmin = Admin.of(bootstrapServers())) {
       topicAdmin.creator().topic(topicName).numberOfPartitions(3).create();
       // wait for syncing topic creation
       TimeUnit.SECONDS.sleep(5);
@@ -264,7 +264,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
   @Test
   void testReplicaSize() throws ExecutionException, InterruptedException {
     var topicName = "testReplicaSize";
-    try (var topicAdmin = TopicAdmin.of(bootstrapServers());
+    try (var topicAdmin = Admin.of(bootstrapServers());
         var producer = Producer.builder().bootstrapServers(bootstrapServers()).build()) {
       producer.sender().topic(topicName).key(new byte[100]).run().toCompletableFuture().get();
       var originSize =
@@ -296,7 +296,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
   @Test
   void testCompact() throws InterruptedException {
     var topicName = "testCompacted";
-    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+    try (var topicAdmin = Admin.of(bootstrapServers())) {
       topicAdmin.creator().topic(topicName).compactionMaxLag(Duration.ofSeconds(1)).create();
 
       var key = "key";
@@ -347,4 +347,13 @@ public class TopicAdminTest extends RequireBrokerCluster {
       }
     }
   }
+
+  @Test
+  void testBrokerConfigs() {
+    try (var admin = Admin.of(bootstrapServers())) {
+      var brokerConfigs = admin.brokers();
+      Assertions.assertEquals(3, brokerConfigs.size());
+      brokerConfigs.values().forEach(c -> Assertions.assertNotEquals(0, c.keys().size()));
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/admin/ReplicaCollieTest.java b/app/src/test/java/org/astraea/admin/ReplicaCollieTest.java
index c9bdabad..2812db52 100644
--- a/app/src/test/java/org/astraea/admin/ReplicaCollieTest.java
+++ b/app/src/test/java/org/astraea/admin/ReplicaCollieTest.java
@@ -35,7 +35,7 @@ public class ReplicaCollieTest extends RequireBrokerCluster {
   @DisabledOnOs(WINDOWS)
   void testBrokerMigrator() throws InterruptedException {
     var topicName = "ReplicaCollieTest-Broker";
-    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+    try (var topicAdmin = Admin.of(bootstrapServers())) {
       topicAdmin
           .creator()
           .topic(topicName)
@@ -77,7 +77,7 @@ public class ReplicaCollieTest extends RequireBrokerCluster {
   @DisabledOnOs(WINDOWS)
   void testPathMigrator() throws InterruptedException {
     var topicName = "ReplicaCollieTest-Path";
-    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+    try (var topicAdmin = Admin.of(bootstrapServers())) {
       topicAdmin
           .creator()
           .topic(topicName)
@@ -120,7 +120,7 @@ public class ReplicaCollieTest extends RequireBrokerCluster {
 
   private void test(boolean verify) throws InterruptedException {
     var topicName = "ReplicaCollieTest-" + verify;
-    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+    try (var topicAdmin = Admin.of(bootstrapServers())) {
       topicAdmin
           .creator()
           .topic(topicName)
diff --git a/app/src/test/java/org/astraea/admin/ReplicaSyncingMonitorIntegrationTest.java b/app/src/test/java/org/astraea/admin/ReplicaSyncingMonitorIntegrationTest.java
index 06e567ff..14f375a1 100644
--- a/app/src/test/java/org/astraea/admin/ReplicaSyncingMonitorIntegrationTest.java
+++ b/app/src/test/java/org/astraea/admin/ReplicaSyncingMonitorIntegrationTest.java
@@ -24,7 +24,7 @@ class ReplicaSyncingMonitorIntegrationTest extends RequireBrokerCluster {
   @Test
   void execute() throws IOException, InterruptedException {
     // arrange
-    try (TopicAdmin topicAdmin = TopicAdmin.of(bootstrapServers())) {
+    try (Admin topicAdmin = Admin.of(bootstrapServers())) {
       topicAdmin
           .creator()
           .topic(TOPIC_NAME)
diff --git a/app/src/test/java/org/astraea/admin/ReplicaSyncingMonitorTest.java b/app/src/test/java/org/astraea/admin/ReplicaSyncingMonitorTest.java
index cf81fa18..f15953d2 100644
--- a/app/src/test/java/org/astraea/admin/ReplicaSyncingMonitorTest.java
+++ b/app/src/test/java/org/astraea/admin/ReplicaSyncingMonitorTest.java
@@ -70,7 +70,7 @@ class ReplicaSyncingMonitorTest {
   void execute() throws InterruptedException {
     // arrange
     int interval = 10;
-    TopicAdmin mockTopicAdmin = mock(TopicAdmin.class);
+    Admin mockTopicAdmin = mock(Admin.class);
     when(mockTopicAdmin.topicNames()).thenReturn(Set.of("topic-1", "topic-2", "topic-3"));
     when(mockTopicAdmin.replicas(anySet()))
         .thenReturn(
@@ -129,7 +129,7 @@ class ReplicaSyncingMonitorTest {
   void executeWithKeepTrack() throws InterruptedException {
     // arrange
     int interval = 10;
-    TopicAdmin mockTopicAdmin = mock(TopicAdmin.class);
+    Admin mockTopicAdmin = mock(Admin.class);
     when(mockTopicAdmin.topicNames()).thenReturn(Set.of("topic-1"));
     when(mockTopicAdmin.replicas(anySet()))
         .thenReturn(
@@ -210,7 +210,7 @@ class ReplicaSyncingMonitorTest {
   void executeWithTopic() {
     // arrange
     int interval = 10;
-    TopicAdmin mockTopicAdmin = mock(TopicAdmin.class);
+    Admin mockTopicAdmin = mock(Admin.class);
     when(mockTopicAdmin.replicas(Set.of("target-topic")))
         .thenReturn(
             Map.of(
@@ -248,7 +248,7 @@ class ReplicaSyncingMonitorTest {
   @Test
   void findNonSyncedTopicPartition() {
     // arrange
-    final TopicAdmin mockTopicAdmin = mock(TopicAdmin.class);
+    final Admin mockTopicAdmin = mock(Admin.class);
     final Set<String> topics = Set.of("topic1", "topic2");
     final List<Replica> replicaList1 =
         List.of(
diff --git a/app/src/test/java/org/astraea/admin/TopicExplorerTest.java b/app/src/test/java/org/astraea/admin/TopicExplorerTest.java
index 032af0db..ed58dc4e 100644
--- a/app/src/test/java/org/astraea/admin/TopicExplorerTest.java
+++ b/app/src/test/java/org/astraea/admin/TopicExplorerTest.java
@@ -86,7 +86,7 @@ public class TopicExplorerTest extends RequireBrokerCluster {
     // arrange
     var topicName = "TopicExplorerTest-testExecute-Topic";
     var groupName = "TopicExplorerTest-testExecute-ConsumerGroup";
-    try (var admin = TopicAdmin.of(bootstrapServers())) {
+    try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).numberOfPartitions(3).numberOfReplicas((short) 1).create();
       consumer(Set.of(topicName), groupName, "alpha");
       consumer(Set.of(topicName), groupName, "beta");
@@ -219,7 +219,7 @@ public class TopicExplorerTest extends RequireBrokerCluster {
                 new Replica(1000, 0, 0, false, false, false, "?"),
                 new Replica(1001, 0, 0, false, false, false, "?"),
                 new Replica(1002, 0, 0, false, false, false, "?")));
-    TopicAdmin mock = Mockito.mock(TopicAdmin.class);
+    Admin mock = Mockito.mock(Admin.class);
 
     Mockito.when(mock.replicas(Set.of(topicName0))).thenReturn(payload0);
     Mockito.when(mock.replicas(Set.of(topicName1))).thenReturn(payload1);
diff --git a/app/src/test/java/org/astraea/cost/topic/GetPartitionInfTest.java b/app/src/test/java/org/astraea/cost/topic/GetPartitionInfTest.java
index 0c4e5a5c..4eb590af 100644
--- a/app/src/test/java/org/astraea/cost/topic/GetPartitionInfTest.java
+++ b/app/src/test/java/org/astraea/cost/topic/GetPartitionInfTest.java
@@ -6,7 +6,7 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
-import org.astraea.admin.TopicAdmin;
+import org.astraea.admin.Admin;
 import org.astraea.producer.Producer;
 import org.astraea.producer.Serializer;
 import org.astraea.service.RequireBrokerCluster;
@@ -14,17 +14,17 @@ import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
 public class GetPartitionInfTest extends RequireBrokerCluster {
-  static TopicAdmin admin;
+  static Admin admin;
 
   @BeforeAll
   static void setup() throws ExecutionException, InterruptedException {
-    admin = TopicAdmin.of(bootstrapServers());
+    admin = Admin.of(bootstrapServers());
     Map<Integer, String> topicName = new HashMap<>();
     topicName.put(0, "testPartitionScore0");
     topicName.put(1, "testPartitionScore1");
     topicName.put(2, "testPartitionScore2");
 
-    try (var admin = TopicAdmin.of(bootstrapServers())) {
+    try (var admin = Admin.of(bootstrapServers())) {
       admin
           .creator()
           .topic(topicName.get(0))
diff --git a/app/src/test/java/org/astraea/cost/topic/PartitionScoreTest.java b/app/src/test/java/org/astraea/cost/topic/PartitionScoreTest.java
index 90c64b9e..3c592247 100644
--- a/app/src/test/java/org/astraea/cost/topic/PartitionScoreTest.java
+++ b/app/src/test/java/org/astraea/cost/topic/PartitionScoreTest.java
@@ -6,7 +6,7 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
-import org.astraea.admin.TopicAdmin;
+import org.astraea.admin.Admin;
 import org.astraea.producer.Producer;
 import org.astraea.producer.Serializer;
 import org.astraea.service.RequireBrokerCluster;
@@ -14,16 +14,16 @@ import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
 public class PartitionScoreTest extends RequireBrokerCluster {
-  static TopicAdmin admin;
+  static Admin admin;
 
   @BeforeAll
   static void setup() throws ExecutionException, InterruptedException {
-    admin = TopicAdmin.of(bootstrapServers());
+    admin = Admin.of(bootstrapServers());
     Map<Integer, String> topicName = new HashMap<>();
     topicName.put(0, "testPartitionScore0");
     topicName.put(1, "testPartitionScore1");
     topicName.put(2, "__consumer_offsets");
-    try (var admin = TopicAdmin.of(bootstrapServers())) {
+    try (var admin = Admin.of(bootstrapServers())) {
       admin
           .creator()
           .topic(topicName.get(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 f6e9d890..85495d94 100644
--- a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java
+++ b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java
@@ -15,7 +15,7 @@ import javax.management.MBeanServer;
 import javax.management.remote.JMXConnectorServer;
 import javax.management.remote.JMXConnectorServerFactory;
 import javax.management.remote.JMXServiceURL;
-import org.astraea.admin.TopicAdmin;
+import org.astraea.admin.Admin;
 import org.astraea.metrics.java.JvmMemory;
 import org.astraea.metrics.java.OperatingSystemInfo;
 import org.astraea.metrics.jmx.MBeanClient;
@@ -123,7 +123,7 @@ class KafkaMetricsTest extends RequireBrokerCluster {
   @Test
   void testSize() throws IOException {
     // arrange
-    try (TopicAdmin admin = TopicAdmin.of(bootstrapServers())) {
+    try (Admin admin = Admin.of(bootstrapServers())) {
       String topicName = getClass().getName();
       admin.creator().topic(topicName).numberOfPartitions(10).create();
 
diff --git a/app/src/test/java/org/astraea/performance/PerformanceTest.java b/app/src/test/java/org/astraea/performance/PerformanceTest.java
index 67a2d732..1fbf799e 100644
--- a/app/src/test/java/org/astraea/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/performance/PerformanceTest.java
@@ -9,7 +9,7 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
 import org.apache.kafka.common.TopicPartition;
-import org.astraea.admin.TopicAdmin;
+import org.astraea.admin.Admin;
 import org.astraea.concurrent.Executor;
 import org.astraea.concurrent.State;
 import org.astraea.consumer.Consumer;
@@ -64,7 +64,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     Assertions.assertEquals(1, producerExecutors.size());
     Assertions.assertFalse(producerExecutors.get(0).transactional());
 
-    try (var admin = TopicAdmin.of(bootstrapServers())) {
+    try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topic).numberOfPartitions(1).create();
       // wait for topic creation
       TimeUnit.SECONDS.sleep(2);
diff --git a/app/src/test/java/org/astraea/performance/ProducerExecutorTest.java b/app/src/test/java/org/astraea/performance/ProducerExecutorTest.java
index 32336b84..af987dc1 100644
--- a/app/src/test/java/org/astraea/performance/ProducerExecutorTest.java
+++ b/app/src/test/java/org/astraea/performance/ProducerExecutorTest.java
@@ -10,7 +10,7 @@ import java.util.function.Supplier;
 import java.util.stream.Stream;
 import org.apache.kafka.common.TopicPartition;
 import org.astraea.Utils;
-import org.astraea.admin.TopicAdmin;
+import org.astraea.admin.Admin;
 import org.astraea.concurrent.State;
 import org.astraea.producer.Producer;
 import org.astraea.service.RequireBrokerCluster;
@@ -27,7 +27,7 @@ public class ProducerExecutorTest extends RequireBrokerCluster {
   void testSpecifiedPartition(ProducerExecutor executor) throws InterruptedException {
     var specifiedPartition = 1;
     ((MyPartitionSupplier) executor.partitionSupplier()).partition = specifiedPartition;
-    try (var admin = TopicAdmin.of(bootstrapServers())) {
+    try (var admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(executor.topic()).numberOfPartitions(specifiedPartition + 1).create();
       // wait for topic creation
       TimeUnit.SECONDS.sleep(2);
diff --git a/app/src/test/java/org/astraea/web/GroupHandlerTest.java b/app/src/test/java/org/astraea/web/GroupHandlerTest.java
index aa5880a9..068212ce 100644
--- a/app/src/test/java/org/astraea/web/GroupHandlerTest.java
+++ b/app/src/test/java/org/astraea/web/GroupHandlerTest.java
@@ -7,7 +7,7 @@ import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import org.astraea.Utils;
-import org.astraea.admin.TopicAdmin;
+import org.astraea.admin.Admin;
 import org.astraea.consumer.Consumer;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
@@ -19,7 +19,7 @@ public class GroupHandlerTest extends RequireBrokerCluster {
   void testListGroups() throws InterruptedException {
     var topicName = Utils.randomString(10);
     var groupId = Utils.randomString(10);
-    try (TopicAdmin admin = TopicAdmin.of(bootstrapServers())) {
+    try (Admin admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).create();
       TimeUnit.SECONDS.sleep(3);
 
@@ -43,7 +43,7 @@ public class GroupHandlerTest extends RequireBrokerCluster {
 
   @Test
   void testQueryNonexistentGroup() {
-    try (TopicAdmin admin = TopicAdmin.of(bootstrapServers())) {
+    try (Admin admin = Admin.of(bootstrapServers())) {
       var handler = new GroupHandler(admin);
       var exception =
           Assertions.assertThrows(
@@ -57,7 +57,7 @@ public class GroupHandlerTest extends RequireBrokerCluster {
   void testQuerySingleGroup() throws InterruptedException {
     var topicName = Utils.randomString(10);
     var groupId = Utils.randomString(10);
-    try (TopicAdmin admin = TopicAdmin.of(bootstrapServers())) {
+    try (Admin admin = Admin.of(bootstrapServers())) {
       var handler = new GroupHandler(admin);
 
       try (var consumer =
diff --git a/app/src/test/java/org/astraea/web/TopicHandlerTest.java b/app/src/test/java/org/astraea/web/TopicHandlerTest.java
index 19cc1e2b..b7b500c6 100644
--- a/app/src/test/java/org/astraea/web/TopicHandlerTest.java
+++ b/app/src/test/java/org/astraea/web/TopicHandlerTest.java
@@ -5,7 +5,7 @@ import java.util.NoSuchElementException;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import org.astraea.Utils;
-import org.astraea.admin.TopicAdmin;
+import org.astraea.admin.Admin;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -15,7 +15,7 @@ public class TopicHandlerTest extends RequireBrokerCluster {
   @Test
   void testListTopics() throws InterruptedException {
     var topicName = Utils.randomString(10);
-    try (TopicAdmin admin = TopicAdmin.of(bootstrapServers())) {
+    try (Admin admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).create();
       TimeUnit.SECONDS.sleep(3);
       var handler = new TopicHandler(admin);
@@ -30,7 +30,7 @@ public class TopicHandlerTest extends RequireBrokerCluster {
 
   @Test
   void testQueryNonexistentTopic() {
-    try (TopicAdmin admin = TopicAdmin.of(bootstrapServers())) {
+    try (Admin admin = Admin.of(bootstrapServers())) {
       var handler = new TopicHandler(admin);
       var exception =
           Assertions.assertThrows(
@@ -43,7 +43,7 @@ public class TopicHandlerTest extends RequireBrokerCluster {
   @Test
   void testQuerySingleTopic() throws InterruptedException {
     var topicName = Utils.randomString(10);
-    try (TopicAdmin admin = TopicAdmin.of(bootstrapServers())) {
+    try (Admin admin = Admin.of(bootstrapServers())) {
       admin.creator().topic(topicName).create();
       TimeUnit.SECONDS.sleep(3);
       var handler = new TopicHandler(admin);