Repository: skiptests/astraea
Pull request #423 (2022-06-24 05:33:55)

Enable to remove all members/static members
這邊新增兩個方法給admin
1. 刪除所有members (static and dynamic)
2. 刪除static members

kafka目前不支援刪除dynamic members :(


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 2038d3ef..66072563 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,15 @@ public interface Admin extends Closeable {
    */
   Map<String, Transaction> transactions(Set<String> transactionIds);
 
+  /** @param groupId to remove all (dynamic and static) members */
+  void removeAllMembers(String groupId);
+
+  /**
+   * @param groupId to remove static members
+   * @param members group instance id (static member)
+   */
+  void removeStaticMembers(String groupId, Set<String> members);
+
   @Override
   void close();
 }
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 27b91bfc..4d622fa3 100644
--- a/app/src/main/java/org/astraea/app/admin/Builder.java
+++ b/app/src/main/java/org/astraea/app/admin/Builder.java
@@ -36,9 +36,11 @@ import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.ConsumerGroupListing;
 import org.apache.kafka.clients.admin.ListTopicsOptions;
 import org.apache.kafka.clients.admin.MemberDescription;
+import org.apache.kafka.clients.admin.MemberToRemove;
 import org.apache.kafka.clients.admin.NewPartitionReassignment;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.RemoveMembersFromConsumerGroupOptions;
 import org.apache.kafka.clients.admin.TransactionListing;
 import org.apache.kafka.common.ElectionType;
 import org.apache.kafka.common.TopicPartitionReplica;
@@ -551,6 +553,32 @@ public class Builder {
                   .collect(
                       Collectors.toMap(Map.Entry::getKey, e -> Transaction.from(e.getValue()))));
     }
+
+    @Override
+    public void removeAllMembers(String groupId) {
+      Utils.packException(
+          () ->
+              admin
+                  .removeMembersFromConsumerGroup(
+                      groupId, new RemoveMembersFromConsumerGroupOptions())
+                  .all()
+                  .get());
+    }
+
+    @Override
+    public void removeStaticMembers(String groupId, Set<String> members) {
+      Utils.packException(
+          () ->
+              admin
+                  .removeMembersFromConsumerGroup(
+                      groupId,
+                      new RemoveMembersFromConsumerGroupOptions(
+                          members.stream()
+                              .map(MemberToRemove::new)
+                              .collect(Collectors.toUnmodifiableList())))
+                  .all()
+                  .get());
+    }
   }
 
   private static class ConfigImpl implements Config {
diff --git a/app/src/main/java/org/astraea/app/consumer/Builder.java b/app/src/main/java/org/astraea/app/consumer/Builder.java
index d618f031..11a13084 100644
--- a/app/src/main/java/org/astraea/app/consumer/Builder.java
+++ b/app/src/main/java/org/astraea/app/consumer/Builder.java
@@ -39,7 +39,8 @@ abstract class Builder<Key, Value> {
    * @return this builder
    */
   public Builder<Key, Value> fromBeginning() {
-    return config(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+    this.configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+    return this;
   }
 
   /**
@@ -48,7 +49,8 @@ abstract class Builder<Key, Value> {
    * @return this builder
    */
   public Builder<Key, Value> fromLatest() {
-    return config(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
+    this.configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
+    return this;
   }
 
   /**
@@ -76,23 +78,15 @@ abstract class Builder<Key, Value> {
     return (Builder<Key, NewValue>) this;
   }
 
-  public Builder<Key, Value> config(String key, String value) {
-    this.configs.put(key, value);
-    return this;
-  }
-
-  public Builder<Key, Value> configs(Map<String, String> configs) {
-    this.configs.putAll(configs);
-    return this;
-  }
-
   public Builder<Key, Value> bootstrapServers(String bootstrapServers) {
-    return config(
+    this.configs.put(
         ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Objects.requireNonNull(bootstrapServers));
+    return this;
   }
 
   public Builder<Key, Value> isolation(Isolation isolation) {
-    return config(ConsumerConfig.ISOLATION_LEVEL_CONFIG, isolation.nameOfKafka());
+    this.configs.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, isolation.nameOfKafka());
+    return this;
   }
 
   /** @return consumer instance. The different builders may return inherited consumer interface. */
diff --git a/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java b/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java
index 3f54e160..3c2d5e72 100644
--- a/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java
+++ b/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java
@@ -79,15 +79,13 @@ public class PartitionsBuilder<Key, Value> extends Builder<Key, Value> {
     return (PartitionsBuilder<Key, NewValue>) super.valueDeserializer(valueDeserializer);
   }
 
-  @Override
   public PartitionsBuilder<Key, Value> config(String key, String value) {
-    super.config(key, value);
+    this.configs.put(key, value);
     return this;
   }
 
-  @Override
   public PartitionsBuilder<Key, Value> configs(Map<String, String> configs) {
-    super.configs(configs);
+    this.configs.putAll(configs);
     return this;
   }
 
diff --git a/app/src/main/java/org/astraea/app/consumer/SubscribedConsumer.java b/app/src/main/java/org/astraea/app/consumer/SubscribedConsumer.java
index d2c05be9..3690f074 100644
--- a/app/src/main/java/org/astraea/app/consumer/SubscribedConsumer.java
+++ b/app/src/main/java/org/astraea/app/consumer/SubscribedConsumer.java
@@ -17,6 +17,7 @@
 package org.astraea.app.consumer;
 
 import java.time.Duration;
+import java.util.Optional;
 
 /**
  * This inherited consumer offers function related to consumer group.
@@ -32,4 +33,13 @@ public interface SubscribedConsumer<Key, Value> extends Consumer<Key, Value> {
    * @param timeout to wait commit.
    */
   void commitOffsets(Duration timeout);
+
+  /** @return the group id including this consumer */
+  String groupId();
+
+  /** @return the member id used by this consumer */
+  String memberId();
+
+  /** @return group instance id (static member) */
+  Optional<String> groupInstanceId();
 }
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 c975fba2..40f900c1 100644
--- a/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
+++ b/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
@@ -21,6 +21,7 @@ import static java.util.Objects.requireNonNull;
 import java.time.Duration;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -31,12 +32,15 @@ public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
 
   TopicsBuilder(Set<String> topics) {
     this.topics = requireNonNull(topics);
-    config(ConsumerConfig.GROUP_ID_CONFIG, "groupId-" + System.currentTimeMillis());
   }
 
   public TopicsBuilder<Key, Value> groupId(String groupId) {
-    config(ConsumerConfig.GROUP_ID_CONFIG, requireNonNull(groupId));
-    return this;
+    return config(ConsumerConfig.GROUP_ID_CONFIG, requireNonNull(groupId));
+  }
+
+  public TopicsBuilder<Key, Value> groupInstanceId(String groupInstanceId) {
+    this.configs.remove(ConsumerConfig.GROUP_ID_CONFIG);
+    return config(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, requireNonNull(groupInstanceId));
   }
 
   public TopicsBuilder<Key, Value> consumerRebalanceListener(ConsumerRebalanceListener listener) {
@@ -91,15 +95,13 @@ public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
     return (TopicsBuilder<Key, NewValue>) super.valueDeserializer(valueDeserializer);
   }
 
-  @Override
   public TopicsBuilder<Key, Value> config(String key, String value) {
-    super.config(key, value);
+    this.configs.put(key, value);
     return this;
   }
 
-  @Override
   public TopicsBuilder<Key, Value> configs(Map<String, String> configs) {
-    super.configs(configs);
+    this.configs.putAll(configs);
     return this;
   }
 
@@ -122,6 +124,9 @@ public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
   @SuppressWarnings("unchecked")
   @Override
   public SubscribedConsumer<Key, Value> build() {
+    // generate group id if it is empty
+    configs.putIfAbsent(ConsumerConfig.GROUP_ID_CONFIG, "groupId-" + System.currentTimeMillis());
+
     var kafkaConsumer =
         new KafkaConsumer<>(
             configs,
@@ -156,5 +161,19 @@ public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
     public void commitOffsets(Duration timeout) {
       kafkaConsumer.commitSync(timeout);
     }
+
+    @Override
+    public String groupId() {
+      return kafkaConsumer.groupMetadata().groupId();
+    }
+
+    @Override
+    public String memberId() {
+      return kafkaConsumer.groupMetadata().memberId();
+    }
+
+    public Optional<String> groupInstanceId() {
+      return kafkaConsumer.groupMetadata().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 1430fdd6..edb0befc 100644
--- a/app/src/test/java/org/astraea/app/admin/AdminTest.java
+++ b/app/src/test/java/org/astraea/app/admin/AdminTest.java
@@ -155,11 +155,6 @@ public class AdminTest extends RequireBrokerCluster {
                 Assertions.assertEquals(0, offset.earliest());
                 Assertions.assertEquals(0, offset.latest());
               });
-
-      admin.creator().topic("a").numberOfPartitions(3).create();
-      // wait for syncing topic creation
-      TimeUnit.SECONDS.sleep(3);
-      Assertions.assertEquals(6, admin.offsets().size());
     }
   }
 
@@ -835,4 +830,60 @@ public class AdminTest extends RequireBrokerCluster {
           transaction.topicPartitions().size());
     }
   }
+
+  @Test
+  void testRemoveAllMembers() {
+    var topicName = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers());
+        var producer = Producer.builder().bootstrapServers(bootstrapServers()).build();
+        var consumer =
+            Consumer.forTopics(Set.of(topicName))
+                .bootstrapServers(bootstrapServers())
+                .fromBeginning()
+                .build()) {
+      producer.sender().topic(topicName).key(new byte[10]).run();
+      producer.flush();
+      Assertions.assertEquals(1, consumer.poll(1, Duration.ofSeconds(5)).size());
+
+      producer.sender().topic(topicName).key(new byte[10]).run();
+      producer.flush();
+      admin.removeAllMembers(consumer.groupId());
+      Assertions.assertEquals(
+          0,
+          admin
+              .consumerGroups(Set.of(consumer.groupId()))
+              .get(consumer.groupId())
+              .activeMembers()
+              .size());
+    }
+  }
+
+  @Test
+  void testRemoveStaticMembers() {
+    var topicName = Utils.randomString(10);
+    var staticId = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers());
+        var producer = Producer.builder().bootstrapServers(bootstrapServers()).build();
+        var consumer =
+            Consumer.forTopics(Set.of(topicName))
+                .bootstrapServers(bootstrapServers())
+                .groupInstanceId(staticId)
+                .fromBeginning()
+                .build()) {
+      producer.sender().topic(topicName).key(new byte[10]).run();
+      producer.flush();
+      Assertions.assertEquals(1, consumer.poll(1, Duration.ofSeconds(5)).size());
+
+      producer.sender().topic(topicName).key(new byte[10]).run();
+      producer.flush();
+      admin.removeStaticMembers(consumer.groupId(), Set.of(consumer.groupInstanceId().get()));
+      Assertions.assertEquals(
+          0,
+          admin
+              .consumerGroups(Set.of(consumer.groupId()))
+              .get(consumer.groupId())
+              .activeMembers()
+              .size());
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java b/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
index d44ba18f..82b74bbf 100644
--- a/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
+++ b/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
@@ -112,8 +112,8 @@ public class ConsumerTest extends RequireBrokerCluster {
 
   @Test
   void testGroupId() {
-    var groupId = "testGroupId";
-    var topic = "testGroupId";
+    var groupId = Utils.randomString(10);
+    var topic = Utils.randomString(10);
     produceData(topic, 1);
 
     java.util.function.BiConsumer<String, Integer> testConsumer =
@@ -126,6 +126,9 @@ public class ConsumerTest extends RequireBrokerCluster {
                   .build()) {
             Assertions.assertEquals(
                 expectedSize, consumer.poll(expectedSize, Duration.ofSeconds(5)).size());
+            Assertions.assertEquals(id, consumer.groupId());
+            Assertions.assertNotNull(consumer.memberId());
+            Assertions.assertFalse(consumer.groupInstanceId().isPresent());
           }
         };
 
@@ -138,6 +141,19 @@ public class ConsumerTest extends RequireBrokerCluster {
     testConsumer.accept("another_group", 0);
   }
 
+  @Test
+  void testGroupInstanceId() {
+    var staticId = Utils.randomString(10);
+    try (var consumer =
+        Consumer.forTopics(Set.of(Utils.randomString(10)))
+            .bootstrapServers(bootstrapServers())
+            .groupInstanceId(staticId)
+            .build()) {
+      Assertions.assertEquals(0, consumer.poll(Duration.ofSeconds(2)).size());
+      Assertions.assertEquals(staticId, consumer.groupInstanceId().get());
+    }
+  }
+
   @Test
   void testDistanceFromLatest() {
     var count = 10;