[COMMON] add assignor and state to ConsumerGroup (#1441)

opensource4you · Jan 16, 2023 · 61dc7c2 · 61dc7c2
1 parent 159fa58
commit 61dc7c2
Show file tree

Hide file tree

Showing 3 changed files with 31 additions and 1 deletion.
diff --git a/common/src/main/java/org/astraea/common/admin/AdminImpl.java b/common/src/main/java/org/astraea/common/admin/AdminImpl.java
@@ -519,6 +519,8 @@ public CompletionStage<List<ConsumerGroup>> consumerGroups(Set<String> consumerG
                     groupId ->
                         new ConsumerGroup(
                             groupId,
+                            consumerGroupDescriptions.get(groupId).partitionAssignor(),
+                            consumerGroupDescriptions.get(groupId).state().name(),
                             NodeInfo.of(consumerGroupDescriptions.get(groupId).coordinator()),
                             consumerGroupMetadata.get(groupId).entrySet().stream()
                                 .collect(

diff --git a/common/src/main/java/org/astraea/common/admin/ConsumerGroup.java b/common/src/main/java/org/astraea/common/admin/ConsumerGroup.java
@@ -23,16 +23,24 @@
 public class ConsumerGroup {
   private final String groupId;
 
+  private final String assignor;
+
+  private final String state;
+
   private final NodeInfo coordinator;
   private final Map<TopicPartition, Long> consumeProgress;
   private final Map<Member, Set<TopicPartition>> assignment;
 
-  public ConsumerGroup(
+  ConsumerGroup(
       String groupId,
+      String assignor,
+      String state,
       NodeInfo coordinator,
       Map<TopicPartition, Long> consumeProgress,
       Map<Member, Set<TopicPartition>> assignment) {
     this.groupId = Objects.requireNonNull(groupId);
+    this.assignor = assignor;
+    this.state = state;
     this.coordinator = coordinator;
     this.consumeProgress = Map.copyOf(consumeProgress);
     this.assignment = Map.copyOf(assignment);
@@ -53,4 +61,12 @@ public Map<Member, Set<TopicPartition>> assignment() {
   public Map<TopicPartition, Long> consumeProgress() {
     return consumeProgress;
   }
+
+  public String assignor() {
+    return assignor;
+  }
+
+  public String state() {
+    return state;
+  }
 }
diff --git a/common/src/test/java/org/astraea/common/admin/AdminTest.java b/common/src/test/java/org/astraea/common/admin/AdminTest.java
@@ -762,6 +762,18 @@ void testConsumerGroups() {
           Assertions.assertEquals(
               1, admin.consumerGroups(Set.of("abc")).toCompletableFuture().join().size());
 
+          admin
+              .consumerGroups(admin.consumerGroupIds().toCompletableFuture().join())
+              .toCompletableFuture()
+              .join()
+              .forEach(
+                  c -> {
+                    Assertions.assertNotNull(c.groupId());
+                    Assertions.assertNotNull(c.coordinator());
+                    Assertions.assertNotNull(c.assignor());
+                    Assertions.assertNotNull(c.state());
+                  });
+
           // test internal
           Assertions.assertNotEquals(
               0,