Repository: skiptests/astraea
Pull request #424 (2022-06-24 06:20:48)

Enable to delete consumer group
這隻PR允許我們去刪除empty group


Patch:
diff --git a/app/src/main/java/org/astraea/app/admin/Admin.java b/app/src/main/java/org/astraea/app/admin/Admin.java
index 66072563..e101b87c 100644
--- a/app/src/main/java/org/astraea/app/admin/Admin.java
+++ b/app/src/main/java/org/astraea/app/admin/Admin.java
@@ -216,6 +216,13 @@ public interface Admin extends Closeable {
    */
   Map<String, Transaction> transactions(Set<String> transactionIds);
 
+  /**
+   * remove an empty group. It causes error if the group has memebrs.
+   *
+   * @param groupId to remove
+   */
+  void removeGroup(String groupId);
+
   /** @param groupId to remove all (dynamic and static) members */
   void removeAllMembers(String groupId);
 
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 4d622fa3..e09c6778 100644
--- a/app/src/main/java/org/astraea/app/admin/Builder.java
+++ b/app/src/main/java/org/astraea/app/admin/Builder.java
@@ -554,6 +554,11 @@ public class Builder {
                       Collectors.toMap(Map.Entry::getKey, e -> Transaction.from(e.getValue()))));
     }
 
+    @Override
+    public void removeGroup(String groupId) {
+      Utils.packException(() -> admin.deleteConsumerGroups(Set.of(groupId)).all().get());
+    }
+
     @Override
     public void removeAllMembers(String groupId) {
       Utils.packException(
diff --git a/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java b/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
index 40f900c1..052c1e6e 100644
--- a/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
+++ b/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
@@ -39,7 +39,6 @@ public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
   }
 
   public TopicsBuilder<Key, Value> groupInstanceId(String groupInstanceId) {
-    this.configs.remove(ConsumerConfig.GROUP_ID_CONFIG);
     return config(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, requireNonNull(groupInstanceId));
   }
 


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 edb0befc..d5bd7d91 100644
--- a/app/src/test/java/org/astraea/app/admin/AdminTest.java
+++ b/app/src/test/java/org/astraea/app/admin/AdminTest.java
@@ -34,6 +34,7 @@ import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.errors.GroupNotEmptyException;
 import org.astraea.app.common.Utils;
 import org.astraea.app.consumer.Consumer;
 import org.astraea.app.consumer.Deserializer;
@@ -886,4 +887,46 @@ public class AdminTest extends RequireBrokerCluster {
               .size());
     }
   }
+
+  @Test
+  void testRemoveGroupWithDynamicMembers() {
+    var groupId = Utils.randomString(10);
+    var topicName = Utils.randomString(10);
+    try (var consumer =
+        Consumer.forTopics(Set.of(topicName))
+            .bootstrapServers(bootstrapServers())
+            .groupId(groupId)
+            .build()) {
+      Assertions.assertEquals(0, consumer.poll(Duration.ofSeconds(3)).size());
+    }
+    try (var admin = Admin.of(bootstrapServers())) {
+      Assertions.assertTrue(admin.consumerGroupIds().contains(groupId));
+      admin.removeGroup(groupId);
+      Assertions.assertFalse(admin.consumerGroupIds().contains(groupId));
+    }
+  }
+
+  @Test
+  void testRemoveGroupWithStaticMembers() throws InterruptedException, ExecutionException {
+    var groupId = Utils.randomString(10);
+    var topicName = Utils.randomString(10);
+    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());
+    }
+
+    try (var admin = Admin.of(bootstrapServers())) {
+      Assertions.assertTrue(admin.consumerGroupIds().contains(groupId));
+      // the static member is existent
+      Assertions.assertThrows(GroupNotEmptyException.class, () -> admin.removeGroup(groupId));
+      // cleanup members
+      admin.removeAllMembers(groupId);
+      admin.removeGroup(groupId);
+      Assertions.assertFalse(admin.consumerGroupIds().contains(groupId));
+    }
+  }
 }