Repository: skiptests/astraea
Pull request #542 (2022-08-02 15:21:09)

Remove limit in RecordHandler#get
resolve https://github.com/skiptests/astraea/issues/441


Patch:
diff --git a/README.md b/README.md
index ffee3952..5ff356f5 100644
--- a/README.md
+++ b/README.md
@@ -7,6 +7,7 @@
 - Zheng-Xian Li <garyparrottt@gmail.com>
 - Xiang-Jun Sun <sean0651101@gmail.com>
 - Zhi-Mao Teng <zhimao.teng@gmail.com>
+- Jia-Sheng Chen <haser1156@gmail.com>
 
 # Kafka Tools
 
diff --git a/app/src/main/java/org/astraea/app/App.java b/app/src/main/java/org/astraea/app/App.java
index c1177822..bb64f1b5 100644
--- a/app/src/main/java/org/astraea/app/App.java
+++ b/app/src/main/java/org/astraea/app/App.java
@@ -23,6 +23,7 @@ import java.util.List;
 import java.util.Map;
 import org.astraea.app.admin.ReplicaSyncingMonitor;
 import org.astraea.app.automation.Automation;
+import org.astraea.app.balancer.BalanceProcessDemo;
 import org.astraea.app.metrics.MetricExplorer;
 import org.astraea.app.performance.Performance;
 import org.astraea.app.web.WebService;
@@ -34,7 +35,8 @@ public class App {
           "performance", Performance.class,
           "monitor", ReplicaSyncingMonitor.class,
           "automation", Automation.class,
-          "web", WebService.class);
+          "web", WebService.class,
+          "balance-demo", BalanceProcessDemo.class);
 
   static void execute(Map<String, Class<?>> mains, List<String> args) throws Throwable {
 
diff --git a/app/src/main/java/org/astraea/app/admin/Builder.java b/app/src/main/java/org/astraea/app/admin/Builder.java
index 6cb7a37f..4921acac 100644
--- a/app/src/main/java/org/astraea/app/admin/Builder.java
+++ b/app/src/main/java/org/astraea/app/admin/Builder.java
@@ -59,6 +59,7 @@ import org.astraea.app.common.Utils;
 public class Builder {
 
   private final Map<String, Object> configs = new HashMap<>();
+  private static final String ERROR_MSG_MEMBER_IS_EMPTY = "leaving members should not be empty";
 
   Builder() {}
 
@@ -555,13 +556,21 @@ public class Builder {
 
     @Override
     public void removeAllMembers(String groupId) {
-      Utils.packException(
-          () ->
+      try {
+        Utils.packException(
+            () -> {
               admin
                   .removeMembersFromConsumerGroup(
                       groupId, new RemoveMembersFromConsumerGroupOptions())
                   .all()
-                  .get());
+                  .get();
+            });
+      } catch (IllegalArgumentException e) {
+        // Deleting all members can't work when there is no members already.
+        if (!ERROR_MSG_MEMBER_IS_EMPTY.equals(e.getMessage())) {
+          throw e;
+        }
+      }
     }
 
     @Override
diff --git a/app/src/main/java/org/astraea/app/balancer/BalanceProcessDemo.java b/app/src/main/java/org/astraea/app/balancer/BalanceProcessDemo.java
new file mode 100644
index 00000000..d83c962a
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/balancer/BalanceProcessDemo.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer;
+
+import com.beust.jcommander.Parameter;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.balancer.executor.RebalanceAdmin;
+import org.astraea.app.balancer.executor.StraightPlanExecutor;
+import org.astraea.app.balancer.generator.ShufflePlanGenerator;
+import org.astraea.app.balancer.log.ClusterLogAllocation;
+
+/**
+ * A simple demo for Balancer, it does the following:
+ *
+ * <ol>
+ *   <li>Generate a random plan for the target cluster.
+ *   <li>Print the information of the generated plan.
+ *   <li>Execute the generated plan against the target cluster.
+ * </ol>
+ */
+public class BalanceProcessDemo {
+
+  public static void main(String[] args) {
+    var argument = org.astraea.app.argument.Argument.parse(new Argument(), args);
+    var rebalancePlanGenerator = new ShufflePlanGenerator(1, 5);
+    var rebalancePlanExecutor = new StraightPlanExecutor();
+    try (var admin = Admin.of(argument.configs())) {
+      Predicate<String> topicFilter = (topic) -> !argument.ignoredTopics.contains(topic);
+      var topics =
+          admin.topicNames().stream().filter(topicFilter).collect(Collectors.toUnmodifiableSet());
+      var clusterInfo = admin.clusterInfo(topics);
+
+      // TODO: implement one interface to select the best plan from many plan ,see #544
+      var rebalancePlan = rebalancePlanGenerator.generate(clusterInfo).findFirst().orElseThrow();
+      System.out.println(rebalancePlan);
+      var targetAllocation =
+          rebalancePlan
+              .rebalancePlan()
+              .orElseThrow(() -> new IllegalStateException("No suitable plan found"));
+
+      System.out.println("[Target Allocation]");
+      System.out.println(ClusterLogAllocation.toString(targetAllocation));
+
+      var rebalanceAdmin = RebalanceAdmin.of(admin, topicFilter);
+      rebalancePlanExecutor.run(rebalanceAdmin, targetAllocation);
+    }
+  }
+
+  public static class Argument extends org.astraea.app.argument.Argument {
+
+    @Parameter(names = {"--ignored.topics"})
+    public Set<String> ignoredTopics =
+        Set.of("__consumer_offsets", "__transaction_state", "__cluster_metadata");
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java b/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
index 482418df..fe96cd54 100644
--- a/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
+++ b/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
@@ -150,7 +150,10 @@ public interface ClusterLogAllocation {
                   .forEach(
                       log ->
                           stringBuilder.append(
-                              String.format("%s(%s) ", log.broker(), log.logDirectory())));
+                              String.format(
+                                  "(%s, %s) ",
+                                  log.broker(),
+                                  log.logDirectory().orElse("log dir not specified"))));
 
               stringBuilder.append(System.lineSeparator());
             });
diff --git a/app/src/main/java/org/astraea/app/web/GroupHandler.java b/app/src/main/java/org/astraea/app/web/GroupHandler.java
index ebc1f1e4..b5499edc 100644
--- a/app/src/main/java/org/astraea/app/web/GroupHandler.java
+++ b/app/src/main/java/org/astraea/app/web/GroupHandler.java
@@ -18,6 +18,7 @@ package org.astraea.app.web;
 
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -27,6 +28,7 @@ import org.astraea.app.admin.TopicPartition;
 public class GroupHandler implements Handler {
   static final String TOPIC_KEY = "topic";
   static final String INSTANCE_KEY = "instance";
+  static final String GROUP_KEY = "group";
   private final Admin admin;
 
   GroupHandler(Admin admin) {
@@ -39,15 +41,25 @@ public class GroupHandler implements Handler {
 
   @Override
   public Response delete(String groupId, Map<String, String> queries) {
-    var groupInstanceId = queries.get(INSTANCE_KEY);
-    var activeMembers = admin.consumerGroups(Set.of(groupId)).get(groupId).activeMembers();
-    // Deleting all members can't work when there is no members already.
-    if (groupInstanceId == null && !activeMembers.isEmpty()) admin.removeAllMembers(groupId);
-    // Deleting nonexistent instance id can cause error
-    if (groupInstanceId != null
-        && activeMembers.stream()
-            .anyMatch(m -> m.groupInstanceId().filter(g -> g.equals(groupInstanceId)).isPresent()))
-      admin.removeStaticMembers(groupId, Set.of(groupInstanceId));
+
+    var shouldDeleteGroup =
+        Optional.ofNullable(queries.get(GROUP_KEY)).filter(Boolean::parseBoolean).isPresent();
+    if (shouldDeleteGroup) {
+      admin.removeAllMembers(groupId);
+      admin.removeGroup(groupId);
+      return Response.OK;
+    }
+
+    var shouldDeleteInstance = Objects.nonNull(queries.get(INSTANCE_KEY));
+    if (shouldDeleteInstance) {
+      var groupInstanceId = queries.get(INSTANCE_KEY);
+      var instanceExisted =
+          admin.consumerGroups(Set.of(groupId)).get(groupId).activeMembers().stream()
+              .anyMatch(x -> x.groupInstanceId().filter(groupInstanceId::equals).isPresent());
+      if (instanceExisted) admin.removeStaticMembers(groupId, Set.of(groupInstanceId));
+    } else {
+      admin.removeAllMembers(groupId);
+    }
     return Response.OK;
   }
 
diff --git a/app/src/main/java/org/astraea/app/web/RecordHandler.java b/app/src/main/java/org/astraea/app/web/RecordHandler.java
index 92e74fee..d43ee1a5 100644
--- a/app/src/main/java/org/astraea/app/web/RecordHandler.java
+++ b/app/src/main/java/org/astraea/app/web/RecordHandler.java
@@ -152,12 +152,7 @@ public class RecordHandler implements Handler {
 
     try (var consumer = consumerBuilder.build()) {
       var limit = Integer.parseInt(queries.getOrDefault(LIMIT, "1"));
-      return new Records(
-          consumer.poll(limit, timeout).stream()
-              .map(Record::new)
-              // TODO: remove limit here (https://github.com/skiptests/astraea/issues/441)
-              .limit(limit)
-              .collect(toList()));
+      return new Records(consumer.poll(limit, timeout).stream().map(Record::new).collect(toList()));
     }
   }
 
diff --git a/docs/web_server/web_api_groups_chinese.md b/docs/web_server/web_api_groups_chinese.md
index 604ba969..b3939cd3 100644
--- a/docs/web_server/web_api_groups_chinese.md
+++ b/docs/web_server/web_api_groups_chinese.md
@@ -111,13 +111,18 @@ DELETE /groups/{groupId}?{instance=id}
 
 參數
 
-| 名稱                       | 說明                                             | 預設  |
-|--------------------------|------------------------------------------------|-----|
-| groupId                    | (必填) 清除此group底下所有的consumer members             | 無   |
-| instance                    | (選填) 只清除此group instance id 關聯的 consumer member | 無   |
+| 名稱                       | 說明                                             | 預設    |
+|--------------------------|------------------------------------------------|-------|
+| groupId                    | (必填) 清除此group底下所有的consumer members             | 無     |
+| group                    | (選填) group=true 將忽視其他參數並刪除group                | 無 |
+| instance                    | (選填) 只清除此group instance id 關聯的 consumer member | 無     |
 
 cURL 範例
-刪除 group-1 底下所有的 consumer members
+
 ```shell
+# 刪除 group-1 底下所有的 consumer members
 curl -X DELETE http://localhost:8001/groups/group-1
+
+# 刪除 group-1
+curl -X DELETE http://localhost:8001/groups/group-1?group=true
 ```
diff --git a/docs/web_server/web_api_records_chinese.md b/docs/web_server/web_api_records_chinese.md
index c8554b44..cddf76e7 100644
--- a/docs/web_server/web_api_records_chinese.md
+++ b/docs/web_server/web_api_records_chinese.md
@@ -113,7 +113,7 @@ GET /records/{topic}
 | partition             | (選填) 指定要讀取之 partition                                                                                        | 無      |
 | keyDeserializer       | (選填) key deserializer                                                                                        | string |
 | valueDeserializer     | (選填) value deserializer                                                                                      | string |
-| limit                 | (選填) 回傳資料筆數上限                                                                                                | 1      |
+| limit                 | (選填) 回傳資料筆數上限，注意此僅為一建議值，您仍有可能取得大於此值之資料筆數                                                                     | 1      |
 | timeout               | (選填) 請求資料最大時限，逾時便中斷本次操作。格式為 `數值` + `單位`，單位可填寫 `days`, `day`, `h`, `m`, `s`, `ms`, `us`, `ns`。範例：1h，5s，1000ms | 5s     |
 | distanceFromLatest    | (選填) 距離最新 offset 往前多少位移量開始拉取資料                                                                               | 無      |
 | distanceFromBeginning | (選填) 距離起始 offset 往後多少位移量開始拉取資料                                                                               | 無      |


Test patch:
diff --git a/app/src/test/java/org/astraea/app/admin/AdminTest.java b/app/src/test/java/org/astraea/app/admin/AdminTest.java
index f7781498..2e8fab32 100644
--- a/app/src/test/java/org/astraea/app/admin/AdminTest.java
+++ b/app/src/test/java/org/astraea/app/admin/AdminTest.java
@@ -890,6 +890,28 @@ public class AdminTest extends RequireBrokerCluster {
     }
   }
 
+  @Test
+  void testRemoveEmptyMember() {
+    var topicName = Utils.randomString(10);
+    var groupId = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers())) {
+      try (var consumer =
+          Consumer.forTopics(Set.of(topicName))
+              .bootstrapServers(bootstrapServers())
+              .groupId(groupId)
+              .groupInstanceId(Utils.randomString(10))
+              .build()) {
+        Assertions.assertEquals(0, consumer.poll(Duration.ofSeconds(3)).size());
+      }
+      Assertions.assertEquals(
+          1, admin.consumerGroups(Set.of(groupId)).get(groupId).activeMembers().size());
+      admin.removeAllMembers(groupId);
+      Assertions.assertEquals(
+          0, admin.consumerGroups(Set.of(groupId)).get(groupId).activeMembers().size());
+      admin.removeAllMembers(groupId);
+    }
+  }
+
   @Test
   void testRemoveStaticMembers() {
     var topicName = Utils.randomString(10);
diff --git a/app/src/test/java/org/astraea/app/balancer/BalanceProcessDemoTest.java b/app/src/test/java/org/astraea/app/balancer/BalanceProcessDemoTest.java
new file mode 100644
index 00000000..47b79a12
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/balancer/BalanceProcessDemoTest.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer;
+
+import java.time.Duration;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.balancer.log.ClusterLogAllocation;
+import org.astraea.app.common.Utils;
+import org.astraea.app.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Test;
+
+public class BalanceProcessDemoTest extends RequireBrokerCluster {
+
+  @Test
+  void run() {
+    // prepare topics
+    createTopics();
+
+    // before operation allocation
+    System.out.println("[Before]");
+    describeCurrentAllocation();
+
+    // run
+    System.out.println("[Execute]");
+    BalanceProcessDemo.main(new String[] {"--bootstrap.servers", bootstrapServers()});
+
+    // after operation allocation
+    System.out.println("[After]");
+    describeCurrentAllocation();
+  }
+
+  void describeCurrentAllocation() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var cla = ClusterLogAllocation.of(admin.clusterInfo());
+      System.out.println(ClusterLogAllocation.toString(cla));
+      System.out.println();
+    }
+  }
+
+  void createTopics() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      admin
+          .creator()
+          .topic("TestingTopic")
+          .numberOfPartitions(10)
+          .numberOfReplicas((short) 2)
+          .create();
+    }
+    Utils.sleep(Duration.ofSeconds(1));
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/web/GroupHandlerTest.java b/app/src/test/java/org/astraea/app/web/GroupHandlerTest.java
index 4294f296..c32ef6a9 100644
--- a/app/src/test/java/org/astraea/app/web/GroupHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/GroupHandlerTest.java
@@ -21,6 +21,8 @@ import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Optional;
 import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.common.Utils;
 import org.astraea.app.consumer.Consumer;
@@ -215,4 +217,51 @@ public class GroupHandlerTest extends RequireBrokerCluster {
       }
     }
   }
+
+  @Test
+  void testDeleteGroup() {
+    var topicName = Utils.randomString(10);
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new GroupHandler(admin);
+
+      var groupIds =
+          IntStream.range(0, 3).mapToObj(x -> Utils.randomString(10)).collect(Collectors.toList());
+      groupIds.forEach(
+          x -> {
+            try (var consumer =
+                Consumer.forTopics(Set.of(topicName))
+                    .bootstrapServers(bootstrapServers())
+                    .groupInstanceId(Utils.randomString(10))
+                    .groupId(x)
+                    .build()) {
+              Assertions.assertEquals(0, consumer.poll(Duration.ofSeconds(3)).size());
+            }
+          });
+
+      var currentGroupIds = admin.consumerGroupIds();
+      Assertions.assertTrue(currentGroupIds.contains(groupIds.get(0)));
+      Assertions.assertTrue(currentGroupIds.contains(groupIds.get(1)));
+      Assertions.assertTrue(currentGroupIds.contains(groupIds.get(2)));
+
+      handler.delete(groupIds.get(2), Map.of());
+      Assertions.assertTrue(admin.consumerGroupIds().contains(groupIds.get(2)));
+
+      handler.delete(groupIds.get(2), Map.of(GroupHandler.GROUP_KEY, "false"));
+      Assertions.assertTrue(admin.consumerGroupIds().contains(groupIds.get(2)));
+
+      handler.delete(groupIds.get(2), Map.of(GroupHandler.GROUP_KEY, "true"));
+      Assertions.assertFalse(admin.consumerGroupIds().contains(groupIds.get(2)));
+
+      var group1Members =
+          admin.consumerGroups(Set.of(groupIds.get(1))).get(groupIds.get(1)).activeMembers();
+      handler.delete(
+          groupIds.get(1),
+          Map.of(
+              GroupHandler.GROUP_KEY,
+              "true",
+              GroupHandler.INSTANCE_KEY,
+              group1Members.get(0).groupInstanceId().get()));
+      Assertions.assertFalse(admin.consumerGroupIds().contains(groupIds.get(1)));
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java b/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
index de7b1c1c..cefc8ba1 100644
--- a/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
@@ -303,9 +303,9 @@ public class RecordHandlerTest extends RequireBrokerCluster {
                 Optional.of(topic),
                 Map.of(DISTANCE_FROM_LATEST, "2", VALUE_DESERIALIZER, "integer")));
 
-    Assertions.assertEquals(1, response.data.size());
+    Assertions.assertEquals(2, response.data.size());
     Assertions.assertEquals(
-        List.of(8), response.data.stream().map(record -> record.value).collect(toList()));
+        List.of(8, 9), response.data.stream().map(record -> record.value).collect(toList()));
   }
 
   @Test
@@ -321,9 +321,9 @@ public class RecordHandlerTest extends RequireBrokerCluster {
                 Optional.of(topic),
                 Map.of(DISTANCE_FROM_BEGINNING, "8", VALUE_DESERIALIZER, "integer")));
 
-    Assertions.assertEquals(1, response.data.size());
+    Assertions.assertEquals(2, response.data.size());
     Assertions.assertEquals(
-        List.of(8), response.data.stream().map(record -> record.value).collect(toList()));
+        List.of(8, 9), response.data.stream().map(record -> record.value).collect(toList()));
   }
 
   @Test
@@ -337,9 +337,9 @@ public class RecordHandlerTest extends RequireBrokerCluster {
             RecordHandler.Records.class,
             handler.get(Optional.of(topic), Map.of(SEEK_TO, "3", VALUE_DESERIALIZER, "integer")));
 
-    Assertions.assertEquals(1, response.data.size());
+    Assertions.assertEquals(2, response.data.size());
     Assertions.assertEquals(
-        List.of(3), response.data.stream().map(record -> record.value).collect(toList()));
+        List.of(3, 4), response.data.stream().map(record -> record.value).collect(toList()));
   }
 
   @Test
@@ -370,8 +370,8 @@ public class RecordHandlerTest extends RequireBrokerCluster {
             RecordHandler.Records.class,
             handler.get(Optional.of(topic), Map.of(DISTANCE_FROM_BEGINNING, "1", PARTITION, "1")));
 
-    Assertions.assertEquals(
-        List.of(1), response.data.stream().map(r -> r.partition).collect(toList()));
+    Assertions.assertTrue(
+        response.data.stream().map(r -> r.partition).filter(p -> p != 1).findAny().isEmpty());
   }
 
   @Test
@@ -387,9 +387,11 @@ public class RecordHandlerTest extends RequireBrokerCluster {
                 Optional.of(topic),
                 Map.of(DISTANCE_FROM_BEGINNING, "2", LIMIT, "3", VALUE_DESERIALIZER, "integer")));
 
-    Assertions.assertEquals(3, response.data.size());
+    // limit is just a recommended size here, we might get more records than limit
+    Assertions.assertEquals(8, response.data.size());
     Assertions.assertEquals(
-        List.of(2, 3, 4), response.data.stream().map(record -> record.value).collect(toList()));
+        List.of(2, 3, 4, 5, 6, 7, 8, 9),
+        response.data.stream().map(record -> record.value).collect(toList()));
   }
 
   @ParameterizedTest