Repository: skiptests/astraea
Pull request #39 (2021-10-06 15:41:59)

Offset tool should show replica details and group details
`
topic='ikea', partition=0, startOffset=407, endOffset=15392375, groups=[Group{id='group-id-1633515512798', offset=10629713}, Group{id='group-id-1633515555181', offset=15390965}, Group{id='group-id-1633515285322', offset=4416648}], replicas=[Replica{broker=59, lag=0, leader=false, inSync=true}, Replica{broker=569, lag=0, leader=true, inSync=true}, Replica{broker=919, lag=0, leader=false, inSync=true}]
`

這隻PR讓我們的offset工具可以顯示跟`partition`有關的所有offsets，包含:

1. earliest offset
2. latest offset
3. replica lag
4. group offset

晚一點會再補上測試


Patch:
diff --git a/app/src/main/java/org/astraea/Utils.java b/app/src/main/java/org/astraea/Utils.java
new file mode 100644
index 00000000..3ea1f2bb
--- /dev/null
+++ b/app/src/main/java/org/astraea/Utils.java
@@ -0,0 +1,18 @@
+package org.astraea;
+
+public final class Utils {
+
+  public static <R> R handleException(Getter<R> getter) {
+    try {
+      return getter.get();
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  public interface Getter<R> {
+    R get() throws Exception;
+  }
+
+  private Utils() {}
+}
diff --git a/app/src/main/java/org/astraea/offset/OffsetExplorer.java b/app/src/main/java/org/astraea/offset/OffsetExplorer.java
index cfe1c51e..e44dd584 100644
--- a/app/src/main/java/org/astraea/offset/OffsetExplorer.java
+++ b/app/src/main/java/org/astraea/offset/OffsetExplorer.java
@@ -1,141 +1,85 @@
 package org.astraea.offset;
 
-import java.io.Closeable;
 import java.io.IOException;
-import java.util.*;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.kafka.clients.CommonClientConfigs;
-import org.apache.kafka.clients.admin.AdminClient;
-import org.apache.kafka.clients.admin.ListTopicsOptions;
-import org.apache.kafka.clients.admin.OffsetSpec;
-import org.apache.kafka.common.TopicPartition;
 import org.astraea.argument.ArgumentUtil;
+import org.astraea.topic.TopicAdmin;
 
 public class OffsetExplorer {
-  interface Admin extends Closeable {
-    Set<String> topics();
 
-    Set<TopicPartition> partitions(Set<String> topics);
-
-    Map<TopicPartition, Long> earliestOffsets(Set<TopicPartition> partitions);
-
-    Map<TopicPartition, Long> latestOffsets(Set<TopicPartition> partitions);
-
-    static Admin of(org.apache.kafka.clients.admin.Admin admin) {
-      return new Admin() {
-        @Override
-        public void close() {
-          admin.close();
-        }
-
-        @Override
-        public Set<String> topics() {
-          try {
-            return admin.listTopics(new ListTopicsOptions().listInternal(true)).names().get();
-          } catch (Exception e) {
-            throw new RuntimeException(e);
-          }
-        }
-
-        @Override
-        public Set<TopicPartition> partitions(Set<String> topics) {
-          try {
-            return admin.describeTopics(topics).all().get().entrySet().stream()
-                .flatMap(
-                    e ->
-                        e.getValue().partitions().stream()
-                            .map(p -> new TopicPartition(e.getKey(), p.partition())))
-                .collect(Collectors.toSet());
-          } catch (Exception e) {
-            throw new RuntimeException(e);
-          }
-        }
-
-        @Override
-        public Map<TopicPartition, Long> earliestOffsets(Set<TopicPartition> partitions) {
-          try {
-            return admin
-                .listOffsets(
-                    partitions.stream()
-                        .collect(Collectors.toMap(e -> e, e -> new OffsetSpec.EarliestSpec())))
-                .all()
-                .get()
-                .entrySet()
-                .stream()
-                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));
-          } catch (Exception e) {
-            throw new RuntimeException(e);
-          }
-        }
+  static class Result {
+    final String topic;
+    final int partition;
+    final long earliestOffset;
+    final long latestOffset;
+    final List<TopicAdmin.Group> groups;
+    final List<TopicAdmin.Replica> replicas;
+
+    Result(
+        String topic,
+        int partition,
+        long startOffset,
+        long endOffset,
+        List<TopicAdmin.Group> groups,
+        List<TopicAdmin.Replica> replicas) {
+      this.topic = topic;
+      this.partition = partition;
+      this.earliestOffset = startOffset;
+      this.latestOffset = endOffset;
+      this.groups = groups;
+      this.replicas = replicas;
+    }
 
-        @Override
-        public Map<TopicPartition, Long> latestOffsets(Set<TopicPartition> partitions) {
-          try {
-            return admin
-                .listOffsets(
-                    partitions.stream()
-                        .collect(Collectors.toMap(e -> e, e -> new OffsetSpec.LatestSpec())))
-                .all()
-                .get()
-                .entrySet()
-                .stream()
-                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));
-          } catch (Exception e) {
-            throw new RuntimeException(e);
-          }
-        }
-      };
+    @Override
+    public String toString() {
+      return "topic='"
+          + topic
+          + '\''
+          + ", partition="
+          + partition
+          + ", earliestOffset="
+          + earliestOffset
+          + ", latestOffset="
+          + latestOffset
+          + ", groups="
+          + groups
+          + ", replicas="
+          + replicas;
     }
   }
 
-  static Map<TopicPartition, Map.Entry<Long, Long>> execute(Admin admin, Set<String> topics) {
-    var topicPartitions = admin.partitions(topics);
-
-    var earliestOffsets = admin.earliestOffsets(topicPartitions);
-    var latestOffsets = admin.latestOffsets(topicPartitions);
-
-    var result = new LinkedHashMap<TopicPartition, Map.Entry<Long, Long>>();
-
-    topics.forEach(
-        topic ->
-            earliestOffsets.entrySet().stream()
-                .filter(e -> e.getKey().topic().equals(topic))
-                .sorted(
-                    Comparator.comparing((Map.Entry<TopicPartition, Long> o) -> o.getKey().topic())
-                        .thenComparingInt(o -> o.getKey().partition()))
-                .forEach(
-                    earliestOffset ->
-                        latestOffsets.entrySet().stream()
-                            .filter(e -> e.getKey().equals(earliestOffset.getKey()))
-                            .forEach(
-                                latestOffset ->
-                                    result.put(
-                                        earliestOffset.getKey(),
-                                        Map.entry(
-                                            earliestOffset.getValue(), latestOffset.getValue())))));
-
-    return result;
+  static List<Result> execute(TopicAdmin admin, Set<String> topics) {
+    var replicas = admin.replicas(topics);
+    var offsets = admin.offset(topics);
+    var groups = admin.groups(topics);
+
+    return replicas.entrySet().stream()
+        .filter(e -> offsets.containsKey(e.getKey()))
+        .map(
+            e ->
+                new Result(
+                    e.getKey().topic(),
+                    e.getKey().partition(),
+                    offsets.get(e.getKey()).earliest,
+                    offsets.get(e.getKey()).latest,
+                    groups.getOrDefault(e.getKey(), List.of()),
+                    e.getValue()))
+        .sorted(
+            Comparator.comparing((Result r) -> r.topic).thenComparing((Result r) -> r.partition))
+        .collect(Collectors.toList());
   }
 
   public static void main(String[] args) throws IOException {
     var argument = ArgumentUtil.parseArgument(new OffsetExplorerArgument(), args);
     try (var admin =
-        Admin.of(
-            AdminClient.create(
-                Map.of(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, argument.brokers)))) {
-      var topics = (argument.topics.isEmpty()) ? admin.topics() : argument.topics;
-      var result = execute(admin, topics);
-      result.forEach(
-          (k, v) ->
-              System.out.println(
-                  "topic: "
-                      + k.topic()
-                      + " partition: "
-                      + k.partition()
-                      + " start: "
-                      + v.getKey()
-                      + " end: "
-                      + v.getValue()));
+        TopicAdmin.of(Map.of(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, argument.brokers))) {
+      execute(admin, argument.topics.isEmpty() ? admin.topics() : argument.topics)
+          .forEach(System.out::println);
     }
   }
 }
diff --git a/app/src/main/java/org/astraea/topic/TopicAdmin.java b/app/src/main/java/org/astraea/topic/TopicAdmin.java
new file mode 100644
index 00000000..5b274883
--- /dev/null
+++ b/app/src/main/java/org/astraea/topic/TopicAdmin.java
@@ -0,0 +1,260 @@
+package org.astraea.topic;
+
+import java.io.Closeable;
+import java.util.*;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.ReplicaInfo;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.astraea.Utils;
+
+public interface TopicAdmin extends Closeable {
+
+  static TopicAdmin of(Map<String, Object> conf) {
+    var admin = Admin.create(conf);
+    return new TopicAdmin() {
+
+      @Override
+      public void close() {
+        admin.close();
+      }
+
+      @Override
+      public Map<TopicPartition, List<Group>> groups(Set<String> topics) {
+        return Utils.handleException(
+                () ->
+                    admin.listConsumerGroups().valid().get().stream()
+                        .map(ConsumerGroupListing::groupId)
+                        .collect(Collectors.toSet()))
+            .stream()
+            .flatMap(
+                group ->
+                    Utils.handleException(
+                            () ->
+                                admin
+                                    .listConsumerGroupOffsets(group)
+                                    .partitionsToOffsetAndMetadata()
+                                    .get())
+                        .entrySet()
+                        .stream()
+                        .filter(e -> topics.contains(e.getKey().topic()))
+                        .map(e -> Map.entry(group, e)))
+            .collect(Collectors.groupingBy(e -> e.getValue().getKey()))
+            .entrySet()
+            .stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    e ->
+                        e.getValue().stream()
+                            .map(
+                                groupOffset ->
+                                    new Group(
+                                        groupOffset.getKey(),
+                                        groupOffset.getValue().getValue().offset()))
+                            .collect(Collectors.toList())));
+      }
+
+      private Map<TopicPartition, Long> earliestOffset(Set<TopicPartition> partitions) {
+
+        return Utils.handleException(
+            () ->
+                admin
+                    .listOffsets(
+                        partitions.stream()
+                            .collect(Collectors.toMap(e -> e, e -> new OffsetSpec.EarliestSpec())))
+                    .all()
+                    .get()
+                    .entrySet()
+                    .stream()
+                    .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset())));
+      }
+
+      private Map<TopicPartition, Long> latestOffset(Set<TopicPartition> partitions) {
+        return Utils.handleException(
+            () ->
+                admin
+                    .listOffsets(
+                        partitions.stream()
+                            .collect(Collectors.toMap(e -> e, e -> new OffsetSpec.LatestSpec())))
+                    .all()
+                    .get()
+                    .entrySet()
+                    .stream()
+                    .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset())));
+      }
+
+      @Override
+      public Set<String> topics() {
+        return Utils.handleException(() -> admin.listTopics().names().get());
+      }
+
+      @Override
+      public Map<TopicPartition, Offset> offset(Set<String> topics) {
+        var partitions =
+            Utils.handleException(
+                () ->
+                    admin.describeTopics(topics).all().get().entrySet().stream()
+                        .flatMap(
+                            e ->
+                                e.getValue().partitions().stream()
+                                    .map(p -> new TopicPartition(e.getKey(), p.partition())))
+                        .collect(Collectors.toSet()));
+        var earliest = earliestOffset(partitions);
+        var latest = latestOffset(partitions);
+        return earliest.entrySet().stream()
+            .filter(e -> latest.containsKey(e.getKey()))
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey, e -> new Offset(e.getValue(), latest.get(e.getKey()))));
+      }
+
+      @Override
+      public Map<TopicPartition, List<Replica>> replicas(Set<String> topics) {
+        var brokers =
+            Utils.handleException(
+                () ->
+                    admin.describeCluster().nodes().get().stream()
+                        .map(Node::id)
+                        .collect(Collectors.toSet()));
+
+        var lags =
+            Utils.handleException(
+                () ->
+                    admin.describeLogDirs(brokers).allDescriptions().get().entrySet().stream()
+                        .collect(
+                            Collectors.toMap(
+                                Map.Entry::getKey,
+                                e ->
+                                    e.getValue().values().stream()
+                                        .flatMap(
+                                            logDirDescription ->
+                                                logDirDescription
+                                                    .replicaInfos()
+                                                    .entrySet()
+                                                    .stream())
+                                        .collect(
+                                            Collectors.toMap(
+                                                Map.Entry::getKey, Map.Entry::getValue)))));
+
+        return Utils.handleException(
+            () ->
+                admin.describeTopics(topics).all().get().entrySet().stream()
+                    .flatMap(
+                        e ->
+                            e.getValue().partitions().stream()
+                                .map(
+                                    topicPartitionInfo ->
+                                        Map.entry(
+                                            new TopicPartition(
+                                                e.getKey(), topicPartitionInfo.partition()),
+                                            topicPartitionInfo.replicas().stream()
+                                                .map(
+                                                    node ->
+                                                        new Replica(
+                                                            node.id(),
+                                                            Optional.ofNullable(
+                                                                    lags.getOrDefault(
+                                                                            node.id(), Map.of())
+                                                                        .get(
+                                                                            new TopicPartition(
+                                                                                e.getKey(),
+                                                                                topicPartitionInfo
+                                                                                    .partition())))
+                                                                .map(ReplicaInfo::offsetLag)
+                                                                .orElse(-1L),
+                                                            topicPartitionInfo.leader().id()
+                                                                == node.id(),
+                                                            topicPartitionInfo
+                                                                .isr()
+                                                                .contains(node)))
+                                                .sorted(
+                                                    Comparator.comparing((Replica r) -> r.broker))
+                                                .collect(Collectors.toList()))))
+                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
+      }
+    };
+  }
+
+  /** @return the topics name (internal topics are excluded) */
+  Set<String> topics();
+
+  /**
+   * @param topics topic names
+   * @return the earliest offset and latest offset for specific topics
+   */
+  Map<TopicPartition, Offset> offset(Set<String> topics);
+
+  /**
+   * @param topics topic names
+   * @return the partition having consumer group id and consumer group offset
+   */
+  Map<TopicPartition, List<Group>> groups(Set<String> topics);
+
+  /**
+   * @param topics topic names
+   * @return the replicas of partition
+   */
+  Map<TopicPartition, List<Replica>> replicas(Set<String> topics);
+
+  class Group {
+    public final String id;
+    public final long offset;
+
+    public Group(String id, long offset) {
+      this.id = id;
+      this.offset = offset;
+    }
+
+    @Override
+    public String toString() {
+      return "Group{" + "id='" + id + '\'' + ", offset=" + offset + '}';
+    }
+  }
+
+  class Offset {
+    public final long earliest;
+    public final long latest;
+
+    public Offset(long earliest, long latest) {
+      this.earliest = earliest;
+      this.latest = latest;
+    }
+
+    @Override
+    public String toString() {
+      return "Offset{" + "earliest=" + earliest + ", latest=" + latest + '}';
+    }
+  }
+
+  class Replica {
+    public final int broker;
+    public final long lag;
+    public final boolean leader;
+    public final boolean inSync;
+
+    public Replica(int broker, long lag, boolean leader, boolean inSync) {
+      this.broker = broker;
+      this.lag = lag;
+      this.leader = leader;
+      this.inSync = inSync;
+    }
+
+    @Override
+    public String toString() {
+      return "Replica{"
+          + "broker="
+          + broker
+          + ", lag="
+          + lag
+          + ", leader="
+          + leader
+          + ", inSync="
+          + inSync
+          + '}';
+    }
+  }
+}


Test patch:
diff --git a/app/src/test/java/org/astraea/offset/OffsetExplorerTest.java b/app/src/test/java/org/astraea/offset/OffsetExplorerTest.java
index 2f8aadae..4d7d0680 100644
--- a/app/src/test/java/org/astraea/offset/OffsetExplorerTest.java
+++ b/app/src/test/java/org/astraea/offset/OffsetExplorerTest.java
@@ -1,52 +1,68 @@
 package org.astraea.offset;
 
-import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import org.apache.kafka.common.TopicPartition;
+import org.astraea.topic.TopicAdmin;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 public class OffsetExplorerTest {
+
   @Test
-  void testExecute() {
+  void test() {
+
     var topicName = "topic";
     var partition = 1000;
+    var topicPartition = new TopicPartition(topicName, partition);
     var latestOffset = 1000L;
     var earliestOffset = 100L;
-
+    var groupId = "thisIsId";
+    var groupOffset = 100;
+    var brokerId = 10;
+    var lag = 2;
+    var leader = true;
+    var inSync = true;
     try (var admin =
-        new OffsetExplorer.Admin() {
-          @Override
-          public void close() {}
-
+        new TopicAdmin() {
           @Override
           public Set<String> topics() {
-            return Collections.singleton("topic");
+            return null;
           }
 
           @Override
-          public Set<TopicPartition> partitions(Set<String> topics) {
-            return Collections.singleton(new TopicPartition(topicName, partition));
+          public Map<TopicPartition, Offset> offset(Set<String> topics) {
+            return Map.of(topicPartition, new Offset(earliestOffset, latestOffset));
           }
 
           @Override
-          public Map<TopicPartition, Long> earliestOffsets(Set<TopicPartition> partitions) {
-            return Map.of(new TopicPartition(topicName, partition), earliestOffset);
+          public Map<TopicPartition, List<Group>> groups(Set<String> topics) {
+            return Map.of(topicPartition, List.of(new Group(groupId, groupOffset)));
           }
 
           @Override
-          public Map<TopicPartition, Long> latestOffsets(Set<TopicPartition> partitions) {
-            return Map.of(new TopicPartition(topicName, partition), latestOffset);
+          public Map<TopicPartition, List<Replica>> replicas(Set<String> topics) {
+            return Map.of(topicPartition, List.of(new Replica(brokerId, lag, leader, inSync)));
           }
+
+          @Override
+          public void close() {}
         }) {
-      var result = OffsetExplorer.execute(admin, Collections.singleton(topicName));
+      var result = OffsetExplorer.execute(admin, Set.of(topicName));
       Assertions.assertEquals(1, result.size());
-      var item = result.entrySet().iterator().next();
-      Assertions.assertEquals(topicName, item.getKey().topic());
-      Assertions.assertEquals(partition, item.getKey().partition());
-      Assertions.assertEquals(earliestOffset, item.getValue().getKey());
-      Assertions.assertEquals(latestOffset, item.getValue().getValue());
+      Assertions.assertEquals(topicName, result.get(0).topic);
+      Assertions.assertEquals(partition, result.get(0).partition);
+      Assertions.assertEquals(earliestOffset, result.get(0).earliestOffset);
+      Assertions.assertEquals(latestOffset, result.get(0).latestOffset);
+      Assertions.assertEquals(1, result.get(0).groups.size());
+      Assertions.assertEquals(groupId, result.get(0).groups.get(0).id);
+      Assertions.assertEquals(groupOffset, result.get(0).groups.get(0).offset);
+      Assertions.assertEquals(1, result.get(0).replicas.size());
+      Assertions.assertEquals(brokerId, result.get(0).replicas.get(0).broker);
+      Assertions.assertEquals(lag, result.get(0).replicas.get(0).lag);
+      Assertions.assertEquals(leader, result.get(0).replicas.get(0).leader);
+      Assertions.assertEquals(inSync, result.get(0).replicas.get(0).inSync);
     }
   }
 }