Repository: skiptests/astraea
Pull request #343 (2022-05-15 13:40:59)

pre-filter the data by user-defined target
新增一個方法來預先過濾資料


Patch:
diff --git a/app/src/main/java/org/astraea/admin/Admin.java b/app/src/main/java/org/astraea/admin/Admin.java
index d6e9dbd2..af51509c 100644
--- a/app/src/main/java/org/astraea/admin/Admin.java
+++ b/app/src/main/java/org/astraea/admin/Admin.java
@@ -24,7 +24,12 @@ public interface Admin extends Closeable {
   Set<String> topicNames();
 
   /** @return the topic name and its configurations. */
-  Map<String, Config> topics();
+  default Map<String, Config> topics() {
+    return topics(topicNames());
+  }
+
+  /** @return the topic name and its configurations. */
+  Map<String, Config> topics(Set<String> topicNames);
 
   /** @return all partitions */
   default Set<TopicPartition> partitions() {
@@ -77,7 +82,15 @@ public interface Admin extends Closeable {
   Map<TopicPartition, List<Replica>> replicas(Set<String> topics);
 
   /** @return all broker id and their configuration */
-  Map<Integer, Config> brokers();
+  default Map<Integer, Config> brokers() {
+    return brokers(brokerIds());
+  }
+
+  /**
+   * @param brokerIds to search
+   * @return broker information
+   */
+  Map<Integer, Config> brokers(Set<Integer> brokerIds);
 
   /** @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 34fbf7ef..4b2921c0 100644
--- a/app/src/main/java/org/astraea/admin/Builder.java
+++ b/app/src/main/java/org/astraea/admin/Builder.java
@@ -206,15 +206,12 @@ public class Builder {
     }
 
     @Override
-    public Map<String, Config> topics() {
-      var topics =
-          Utils.handleException(
-              () -> admin.listTopics(new ListTopicsOptions().listInternal(true)).names().get());
+    public Map<String, Config> topics(Set<String> topicNames) {
       return Utils.handleException(
               () ->
                   admin
                       .describeConfigs(
-                          topics.stream()
+                          topicNames.stream()
                               .map(topic -> new ConfigResource(ConfigResource.Type.TOPIC, topic))
                               .collect(Collectors.toList()))
                       .all()
@@ -231,8 +228,7 @@ public class Builder {
     }
 
     @Override
-    public Map<Integer, Config> brokers() {
-      var brokerIds = brokerIds();
+    public Map<Integer, Config> brokers(Set<Integer> brokerIds) {
       return Utils.handleException(
               () ->
                   admin
diff --git a/app/src/main/java/org/astraea/web/BrokerHandler.java b/app/src/main/java/org/astraea/web/BrokerHandler.java
index 6084eda2..51bad7da 100644
--- a/app/src/main/java/org/astraea/web/BrokerHandler.java
+++ b/app/src/main/java/org/astraea/web/BrokerHandler.java
@@ -4,7 +4,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Optional;
-import java.util.function.Predicate;
+import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 import org.astraea.admin.Admin;
@@ -18,29 +18,22 @@ class BrokerHandler implements Handler {
     this.admin = admin;
   }
 
-  private static void mustBeNumber(Optional<String> target) {
+  Set<Integer> brokers(Optional<String> target) {
     try {
-      target.ifPresent(Integer::valueOf);
+      return Handler.compare(admin.brokerIds(), target.map(Integer::valueOf));
     } catch (NumberFormatException e) {
-      throw new NoSuchElementException("broker: " + target.get() + " does not exist");
+      throw new NoSuchElementException("the broker id must be number");
     }
   }
 
   @Override
   public JsonObject get(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)
+        admin.brokers(brokers(target)).entrySet().stream()
             .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);
+    return new Brokers(brokers);
   }
 
   static class Broker implements JsonObject {
diff --git a/app/src/main/java/org/astraea/web/ErrorObject.java b/app/src/main/java/org/astraea/web/ErrorObject.java
index 585d4e3f..834e678e 100644
--- a/app/src/main/java/org/astraea/web/ErrorObject.java
+++ b/app/src/main/java/org/astraea/web/ErrorObject.java
@@ -12,12 +12,17 @@ class ErrorObject implements JsonObject {
   final String message;
 
   ErrorObject(Exception exception) {
-    this.code = exception instanceof NoSuchElementException ? 404 : 400;
-    this.message = exception.getMessage();
+    this(code(exception), exception.getMessage());
   }
 
   ErrorObject(int code, String message) {
     this.code = code;
     this.message = message;
   }
+
+  private static int code(Exception exception) {
+    if (exception instanceof IllegalArgumentException) return 400;
+    if (exception instanceof NoSuchElementException) return 404;
+    return 400;
+  }
 }
diff --git a/app/src/main/java/org/astraea/web/GroupHandler.java b/app/src/main/java/org/astraea/web/GroupHandler.java
index 026e0ede..f3b33206 100644
--- a/app/src/main/java/org/astraea/web/GroupHandler.java
+++ b/app/src/main/java/org/astraea/web/GroupHandler.java
@@ -2,9 +2,8 @@ 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.Set;
 import java.util.stream.Collectors;
 import org.astraea.admin.Admin;
 
@@ -16,17 +15,18 @@ public class GroupHandler implements Handler {
     this.admin = admin;
   }
 
+  Set<String> groupIds(Optional<String> target) {
+    return Handler.compare(admin.consumerGroupIds(), target);
+  }
+
   @Override
   public JsonObject get(Optional<String> target, Map<String, String> queries) {
-    Predicate<Map.Entry<String, ?>> groupFilter =
-        e -> target.stream().allMatch(t -> t.equals(e.getKey()));
     var topics = admin.topicNames();
-    var consumerGroups = admin.consumerGroups();
+    var consumerGroups = admin.consumerGroups(groupIds(target));
     var offsets = admin.offsets(topics);
 
     var groups =
         consumerGroups.entrySet().stream()
-            .filter(groupFilter)
             .map(
                 cgAndTp ->
                     new Group(
@@ -67,9 +67,7 @@ public class GroupHandler implements Handler {
             .collect(Collectors.toUnmodifiableList());
 
     if (target.isPresent() && groups.size() == 1) return groups.get(0);
-    else if (target.isPresent())
-      throw new NoSuchElementException("group: " + target.get() + " does not exist");
-    else return new Groups(groups);
+    return new Groups(groups);
   }
 
   static class OffsetProgress implements JsonObject {
diff --git a/app/src/main/java/org/astraea/web/Handler.java b/app/src/main/java/org/astraea/web/Handler.java
index 2cc608aa..2f3f280e 100644
--- a/app/src/main/java/org/astraea/web/Handler.java
+++ b/app/src/main/java/org/astraea/web/Handler.java
@@ -8,11 +8,20 @@ import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.Locale;
 import java.util.Map;
+import java.util.NoSuchElementException;
 import java.util.Optional;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 interface Handler extends HttpHandler {
 
+  static <T> Set<T> compare(Set<T> all, Optional<T> target) {
+    var nonexistent = target.stream().filter(id -> !all.contains(id)).collect(Collectors.toSet());
+    if (!nonexistent.isEmpty())
+      throw new NoSuchElementException("[" + nonexistent + "] does not exist");
+    return target.map(Set::of).orElse(all);
+  }
+
   default JsonObject process(HttpExchange exchange) {
     var method = exchange.getRequestMethod().toUpperCase(Locale.ROOT);
     var start = System.currentTimeMillis();
diff --git a/app/src/main/java/org/astraea/web/TopicHandler.java b/app/src/main/java/org/astraea/web/TopicHandler.java
index d143711b..210123e0 100644
--- a/app/src/main/java/org/astraea/web/TopicHandler.java
+++ b/app/src/main/java/org/astraea/web/TopicHandler.java
@@ -3,9 +3,8 @@ package org.astraea.web;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
-import java.util.NoSuchElementException;
 import java.util.Optional;
-import java.util.function.Predicate;
+import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 import org.astraea.admin.Admin;
@@ -19,11 +18,13 @@ class TopicHandler implements Handler {
     this.admin = admin;
   }
 
+  Set<String> topicNames(Optional<String> target) {
+    return Handler.compare(admin.topicNames(), target);
+  }
+
   @Override
   public JsonObject get(Optional<String> target, Map<String, String> queries) {
-    Predicate<Map.Entry<String, ?>> topicFilter =
-        e -> target.stream().allMatch(t -> t.equals(e.getKey()));
-    var topics = admin.topics();
+    var topics = admin.topics(topicNames(target));
     var replicas = admin.replicas(topics.keySet());
     var partitions =
         admin.offsets(topics.keySet()).entrySet().stream()
@@ -43,14 +44,11 @@ class TopicHandler implements Handler {
 
     var topicInfos =
         topics.entrySet().stream()
-            .filter(topicFilter)
             .map(p -> new TopicInfo(p.getKey(), partitions.get(p.getKey()), p.getValue()))
             .collect(Collectors.toUnmodifiableList());
 
     if (target.isPresent() && topicInfos.size() == 1) return topicInfos.get(0);
-    else if (target.isPresent())
-      throw new NoSuchElementException("topic: " + target.get() + " does not exist");
-    else return new Topics(topicInfos);
+    return new Topics(topicInfos);
   }
 
   static class Topics implements JsonObject {


Test patch:
diff --git a/app/src/test/java/org/astraea/admin/AdminTest.java b/app/src/test/java/org/astraea/admin/AdminTest.java
index 17b48a89..2f5b57d7 100644
--- a/app/src/test/java/org/astraea/admin/AdminTest.java
+++ b/app/src/test/java/org/astraea/admin/AdminTest.java
@@ -326,6 +326,7 @@ public class AdminTest extends RequireBrokerCluster {
       var brokerConfigs = admin.brokers();
       Assertions.assertEquals(3, brokerConfigs.size());
       brokerConfigs.values().forEach(c -> Assertions.assertNotEquals(0, c.keys().size()));
+      Assertions.assertEquals(1, admin.brokers(Set.of(brokerIds().iterator().next())).size());
     }
   }
 
diff --git a/app/src/test/java/org/astraea/web/BrokerHandlerTest.java b/app/src/test/java/org/astraea/web/BrokerHandlerTest.java
index 5867c9c2..7d145f16 100644
--- a/app/src/test/java/org/astraea/web/BrokerHandlerTest.java
+++ b/app/src/test/java/org/astraea/web/BrokerHandlerTest.java
@@ -3,6 +3,7 @@ package org.astraea.web;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Optional;
+import java.util.Set;
 import org.astraea.admin.Admin;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
@@ -30,10 +31,8 @@ public class BrokerHandlerTest extends RequireBrokerCluster {
   void testQueryNonexistentBroker() {
     try (Admin admin = Admin.of(bootstrapServers())) {
       var handler = new BrokerHandler(admin);
-      var exception =
-          Assertions.assertThrows(
-              NoSuchElementException.class, () -> handler.get(Optional.of("99999"), Map.of()));
-      Assertions.assertTrue(exception.getMessage().contains("99999"));
+      Assertions.assertThrows(
+          NoSuchElementException.class, () -> handler.get(Optional.of("99999"), Map.of()));
     }
   }
 
@@ -41,15 +40,13 @@ public class BrokerHandlerTest extends RequireBrokerCluster {
   void testQueryInvalidBroker() {
     try (Admin admin = Admin.of(bootstrapServers())) {
       var handler = new BrokerHandler(admin);
-      var exception =
-          Assertions.assertThrows(
-              NoSuchElementException.class, () -> handler.get(Optional.of("abc"), Map.of()));
-      Assertions.assertTrue(exception.getMessage().contains("abc"));
+      Assertions.assertThrows(
+          NoSuchElementException.class, () -> handler.get(Optional.of("abc"), Map.of()));
     }
   }
 
   @Test
-  void testQuerySingleBroker() throws InterruptedException {
+  void testQuerySingleBroker() {
     var brokerId = brokerIds().iterator().next();
     try (Admin admin = Admin.of(bootstrapServers())) {
       var handler = new BrokerHandler(admin);
@@ -61,4 +58,17 @@ public class BrokerHandlerTest extends RequireBrokerCluster {
       Assertions.assertNotEquals(0, broker.configs.size());
     }
   }
+
+  @Test
+  void testBrokers() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new BrokerHandler(admin);
+      Assertions.assertEquals(
+          Set.of(brokerIds().iterator().next()),
+          handler.brokers(Optional.of(String.valueOf(brokerIds().iterator().next()))));
+      Assertions.assertEquals(brokerIds(), handler.brokers(Optional.empty()));
+      Assertions.assertThrows(
+          NoSuchElementException.class, () -> handler.brokers(Optional.of("aaa")));
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/web/GroupHandlerTest.java b/app/src/test/java/org/astraea/web/GroupHandlerTest.java
index c840f76b..4e442290 100644
--- a/app/src/test/java/org/astraea/web/GroupHandlerTest.java
+++ b/app/src/test/java/org/astraea/web/GroupHandlerTest.java
@@ -51,15 +51,13 @@ public class GroupHandlerTest extends RequireBrokerCluster {
   void testQueryNonexistentGroup() {
     try (Admin admin = Admin.of(bootstrapServers())) {
       var handler = new GroupHandler(admin);
-      var exception =
-          Assertions.assertThrows(
-              NoSuchElementException.class, () -> handler.get(Optional.of("unknown"), Map.of()));
-      Assertions.assertTrue(exception.getMessage().contains("unknown"));
+      Assertions.assertThrows(
+          NoSuchElementException.class, () -> handler.get(Optional.of("unknown"), Map.of()));
     }
   }
 
   @Test
-  void testQuerySingleGroup() throws InterruptedException {
+  void testQuerySingleGroup() {
     var topicName = Utils.randomString(10);
     var groupId = Utils.randomString(10);
     try (Admin admin = Admin.of(bootstrapServers())) {
@@ -80,4 +78,27 @@ public class GroupHandlerTest extends RequireBrokerCluster {
       }
     }
   }
+
+  @Test
+  void testGroups() {
+    var topicName = Utils.randomString(10);
+    var groupId = Utils.randomString(10);
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new GroupHandler(admin);
+
+      try (var consumer =
+          Consumer.builder()
+              .groupId(groupId)
+              .topics(Set.of(topicName))
+              .bootstrapServers(bootstrapServers())
+              .build()) {
+        Assertions.assertEquals(0, consumer.poll(Duration.ofSeconds(3)).size());
+        Assertions.assertEquals(Set.of(groupId), handler.groupIds(Optional.of(groupId)));
+        Assertions.assertThrows(
+            NoSuchElementException.class,
+            () -> handler.groupIds(Optional.of(Utils.randomString(10))));
+        Assertions.assertTrue(handler.groupIds(Optional.empty()).contains(groupId));
+      }
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/web/TopicHandlerTest.java b/app/src/test/java/org/astraea/web/TopicHandlerTest.java
index d324eccf..0c563351 100644
--- a/app/src/test/java/org/astraea/web/TopicHandlerTest.java
+++ b/app/src/test/java/org/astraea/web/TopicHandlerTest.java
@@ -3,6 +3,7 @@ package org.astraea.web;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import org.astraea.Utils;
 import org.astraea.admin.Admin;
@@ -32,10 +33,8 @@ public class TopicHandlerTest extends RequireBrokerCluster {
   void testQueryNonexistentTopic() {
     try (Admin admin = Admin.of(bootstrapServers())) {
       var handler = new TopicHandler(admin);
-      var exception =
-          Assertions.assertThrows(
-              NoSuchElementException.class, () -> handler.get(Optional.of("unknown"), Map.of()));
-      Assertions.assertTrue(exception.getMessage().contains("unknown"));
+      Assertions.assertThrows(
+          NoSuchElementException.class, () -> handler.get(Optional.of("unknown"), Map.of()));
     }
   }
 
@@ -53,4 +52,19 @@ public class TopicHandlerTest extends RequireBrokerCluster {
       Assertions.assertNotEquals(0, topicInfo.configs.size());
     }
   }
+
+  @Test
+  void testTopics() throws InterruptedException {
+    var topicName = Utils.randomString(10);
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      admin.creator().topic(topicName).create();
+      TimeUnit.SECONDS.sleep(3);
+      var handler = new TopicHandler(admin);
+      Assertions.assertEquals(Set.of(topicName), handler.topicNames(Optional.of(topicName)));
+      Assertions.assertThrows(
+          NoSuchElementException.class,
+          () -> handler.topicNames(Optional.of(Utils.randomString(10))));
+      Assertions.assertTrue(handler.topicNames(Optional.empty()).contains(topicName));
+    }
+  }
 }