Repository: skiptests/astraea
Pull request #534 (2022-07-31 17:29:25)

Balance Process Demo
@chia7712 這是上個禮拜三提到，要我放上來的 Generator 和 Executor 使用範例

1. `BalancerProcessDemo#main` 是主要的進入點
2. `BalancerProcessDemoTest#run` 提供一個快速在測試內執行的範例

 `BalancerProcessDemoTest#run` 結果:

```
[Before]
[TestingTopic-0] 0(Optional[/tmp/local_kafka14626902519260956202]) 2(Optional[/tmp/local_kafka11253759099362447978]) 
[TestingTopic-1] 2(Optional[/tmp/local_kafka11253759099362447978]) 1(Optional[/tmp/local_kafka16293464009788394727]) 
[TestingTopic-2] 1(Optional[/tmp/local_kafka13005076731925113899]) 0(Optional[/tmp/local_kafka14626902519260956202]) 
[TestingTopic-3] 0(Optional[/tmp/local_kafka16325439987434116715]) 1(Optional[/tmp/local_kafka3614035964014968359]) 
[TestingTopic-4] 2(Optional[/tmp/local_kafka16417985240373605211]) 0(Optional[/tmp/local_kafka17537739788417371504]) 
[TestingTopic-5] 1(Optional[/tmp/local_kafka16293464009788394727]) 2(Optional[/tmp/local_kafka11801277469713354649]) 
[TestingTopic-6] 0(Optional[/tmp/local_kafka17537739788417371504]) 2(Optional[/tmp/local_kafka16417985240373605211]) 
[TestingTopic-7] 2(Optional[/tmp/local_kafka11801277469713354649]) 1(Optional[/tmp/local_kafka3614035964014968359]) 
[TestingTopic-8] 1(Optional[/tmp/local_kafka3614035964014968359]) 0(Optional[/tmp/local_kafka16325439987434116715]) 
[TestingTopic-9] 0(Optional[/tmp/local_kafka14626902519260956202]) 1(Optional[/tmp/local_kafka13005076731925113899]) 


[Execute]
[RebalancePlanProposal]
  Info:
    * Make 2 shuffles.
    * Change replica set of topic TestingTopic partition 2, from 0 to 2 at /tmp/local_kafka11801277469713354649.
    * Change the log identity of topic TestingTopic partition 2 replica at broker 2, from follower to leader

[After]
[TestingTopic-0] 0(Optional[/tmp/local_kafka14626902519260956202]) 2(Optional[/tmp/local_kafka11253759099362447978]) 
[TestingTopic-1] 2(Optional[/tmp/local_kafka11253759099362447978]) 1(Optional[/tmp/local_kafka16293464009788394727]) 
[TestingTopic-2] 2(Optional[/tmp/local_kafka11801277469713354649]) 1(Optional[/tmp/local_kafka13005076731925113899]) 
[TestingTopic-3] 0(Optional[/tmp/local_kafka16325439987434116715]) 1(Optional[/tmp/local_kafka3614035964014968359]) 
[TestingTopic-4] 2(Optional[/tmp/local_kafka16417985240373605211]) 0(Optional[/tmp/local_kafka17537739788417371504]) 
[TestingTopic-5] 1(Optional[/tmp/local_kafka16293464009788394727]) 2(Optional[/tmp/local_kafka11801277469713354649]) 
[TestingTopic-6] 0(Optional[/tmp/local_kafka17537739788417371504]) 2(Optional[/tmp/local_kafka16417985240373605211]) 
[TestingTopic-7] 2(Optional[/tmp/local_kafka11801277469713354649]) 1(Optional[/tmp/local_kafka3614035964014968359]) 
[TestingTopic-8] 1(Optional[/tmp/local_kafka3614035964014968359]) 0(Optional[/tmp/local_kafka16325439987434116715]) 
[TestingTopic-9] 0(Optional[/tmp/local_kafka14626902519260956202]) 1(Optional[/tmp/local_kafka13005076731925113899]) 

```


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/common/Utils.java b/app/src/main/java/org/astraea/app/common/Utils.java
index c8f0fe3b..bc770546 100644
--- a/app/src/main/java/org/astraea/app/common/Utils.java
+++ b/app/src/main/java/org/astraea/app/common/Utils.java
@@ -17,6 +17,7 @@
 package org.astraea.app.common;
 
 import java.io.File;
+import java.lang.reflect.Constructor;
 import java.net.InetAddress;
 import java.net.ServerSocket;
 import java.nio.file.Files;
@@ -34,6 +35,8 @@ import java.util.function.Supplier;
 import java.util.stream.Collector;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import org.astraea.app.cost.CostFunction;
+import org.astraea.app.partitioner.Configuration;
 
 public final class Utils {
 
@@ -253,5 +256,23 @@ public final class Utils {
         .thenApply(f -> futures.stream().map(CompletableFuture::join).collect(Collectors.toList()));
   }
 
+  public static CostFunction constructCostFunction(
+      Class<? extends CostFunction> costClass, Configuration configuration) {
+    for (Constructor<?> constructor : costClass.getConstructors()) {
+      Class<?>[] types = constructor.getParameterTypes();
+      if (types.length == 1 && types[0].isAssignableFrom(Configuration.class)) {
+        return (CostFunction) Utils.packException(() -> constructor.newInstance(configuration));
+      }
+    }
+    for (Constructor<?> constructor : costClass.getConstructors()) {
+      Class<?>[] types = constructor.getParameterTypes();
+      if (types.length == 0) {
+        return (CostFunction) Utils.packException(() -> constructor.newInstance());
+      }
+    }
+    throw new IllegalArgumentException(
+        "No suitable constructor found for class " + costClass.getName());
+  }
+
   private Utils() {}
 }
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/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
 ```


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/common/UtilsTest.java b/app/src/test/java/org/astraea/app/common/UtilsTest.java
index 9ff5b886..47db461d 100644
--- a/app/src/test/java/org/astraea/app/common/UtilsTest.java
+++ b/app/src/test/java/org/astraea/app/common/UtilsTest.java
@@ -24,8 +24,12 @@ import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
+import org.astraea.app.cost.CostFunction;
+import org.astraea.app.partitioner.Configuration;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 public class UtilsTest {
 
@@ -82,4 +86,41 @@ public class UtilsTest {
     Assertions.assertThrows(IllegalArgumentException.class, () -> Utils.requireNonEmpty(""));
     Assertions.assertThrows(NullPointerException.class, () -> Utils.requireNonEmpty(null));
   }
+
+  public static class TestConfigCostFunction implements CostFunction {
+    public TestConfigCostFunction(Configuration configuration) {}
+  }
+
+  public static class TestCostFunction implements CostFunction {
+    public TestCostFunction() {}
+  }
+
+  public static class TestBadCostFunction implements CostFunction {
+    public TestBadCostFunction(int value) {}
+  }
+
+  @ParameterizedTest
+  @ValueSource(classes = {TestCostFunction.class, TestConfigCostFunction.class})
+  void constructCostFunction(Class<? extends CostFunction> aClass) {
+    // arrange
+    var config = Configuration.of(Map.of());
+
+    // act
+    var costFunction = Utils.constructCostFunction(aClass, config);
+
+    // assert
+    Assertions.assertInstanceOf(CostFunction.class, costFunction);
+    Assertions.assertInstanceOf(aClass, costFunction);
+  }
+
+  @Test
+  void constructCostFunctionException() {
+    // arrange
+    var aClass = TestBadCostFunction.class;
+    var config = Configuration.of(Map.of());
+
+    // act, assert
+    Assertions.assertThrows(
+        IllegalArgumentException.class, () -> Utils.constructCostFunction(aClass, config));
+  }
 }
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)));
+    }
+  }
 }