KAFKA-18288: Add support kafka-streams-groups.sh --describe (#19433)

Implement `--describe` and its options: (`--state`, -`-offset`,
`--members` and the combination of them with `--verbose`) as described
in `KIP-1071`.

Reviewers: Lucas Brutschy <[email protected]>, PoAn Yang
<[email protected]>

Author: aliehsaeedii

Diff for: tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java

@@ -18,18 +18,30 @@
 
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.DescribeStreamsGroupsResult;
 import org.apache.kafka.clients.admin.GroupListing;
+import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsSpec;
 import org.apache.kafka.clients.admin.ListGroupsOptions;
 import org.apache.kafka.clients.admin.ListGroupsResult;
+import org.apache.kafka.clients.admin.ListOffsetsResult;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.StreamsGroupDescription;
+import org.apache.kafka.clients.admin.StreamsGroupMemberAssignment;
+import org.apache.kafka.clients.admin.StreamsGroupMemberDescription;
+import org.apache.kafka.clients.admin.StreamsGroupSubtopologyDescription;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.GroupState;
 import org.apache.kafka.common.GroupType;
+import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.server.util.CommandLineUtils;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -49,9 +61,9 @@ public static void main(String[] args) {
             opts.checkArgs();
 
             // should have exactly one action
-            long numberOfActions = Stream.of(opts.listOpt).filter(opts.options::has).count();
+            long numberOfActions = Stream.of(opts.listOpt, opts.describeOpt).filter(opts.options::has).count();
             if (numberOfActions != 1)
-                CommandLineUtils.printUsageAndExit(opts.parser, "Command must include exactly one action: --list.");
+                CommandLineUtils.printUsageAndExit(opts.parser, "Command must include exactly one action: --list, or --describe.");
 
             run(opts);
         } catch (OptionException e) {
@@ -63,6 +75,8 @@ public static void run(StreamsGroupCommandOptions opts) {
         try (StreamsGroupService streamsGroupService = new StreamsGroupService(opts, Map.of())) {
             if (opts.options.has(opts.listOpt)) {
                 streamsGroupService.listGroups();
+            } else if (opts.options.has(opts.describeOpt)) {
+                streamsGroupService.describeGroups();
             } else {
                 throw new IllegalArgumentException("Unknown action!");
             }
@@ -79,7 +93,7 @@ static Set<GroupState> groupStatesFromString(String input) {
         Set<GroupState> validStates = GroupState.groupStatesForType(GroupType.STREAMS);
         if (!validStates.containsAll(parsedStates)) {
             throw new IllegalArgumentException("Invalid state list '" + input + "'. Valid states are: " +
-                    validStates.stream().map(GroupState::toString).collect(Collectors.joining(", ")));
+                validStates.stream().map(GroupState::toString).collect(Collectors.joining(", ")));
         }
         return parsedStates;
     }
@@ -154,6 +168,224 @@ private void printGroupInfo(List<GroupListing> groups) {
             }
         }
 
+        public void describeGroups() throws ExecutionException, InterruptedException {
+            List<String> groups = listStreamsGroups();
+            if (!groups.isEmpty()) {
+                StreamsGroupDescription description = getDescribeGroup(groups.get(0));
+                if (description == null)
+                    return;
+                boolean verbose = opts.options.has(opts.verboseOpt);
+                if (opts.options.has(opts.membersOpt)) {
+                    printMembers(description, verbose);
+                } else if (opts.options.has(opts.stateOpt)) {
+                    printStates(description, verbose);
+                } else {
+                    printOffsets(description, verbose);
+                }
+            }
+        }
+
+        StreamsGroupDescription getDescribeGroup(String group) throws ExecutionException, InterruptedException {
+            DescribeStreamsGroupsResult result = adminClient.describeStreamsGroups(List.of(group));
+            Map<String, StreamsGroupDescription> descriptionMap = result.all().get();
+            return descriptionMap.get(group);
+        }
+
+        private void printMembers(StreamsGroupDescription description, boolean verbose) {
+            final int groupLen = Math.max(15, description.groupId().length());
+            int maxMemberIdLen = 15, maxHostLen = 15, maxClientIdLen = 15;
+            Collection<StreamsGroupMemberDescription> members = description.members();
+            if (isGroupStateValid(description.groupState(), description.members().size())) {
+                maybePrintEmptyGroupState(description.groupId(), description.groupState());
+                for (StreamsGroupMemberDescription member : members) {
+                    maxMemberIdLen = Math.max(maxMemberIdLen, member.memberId().length());
+                    maxHostLen = Math.max(maxHostLen, member.processId().length());
+                    maxClientIdLen = Math.max(maxClientIdLen, member.clientId().length());
+                }
+
+                if (!verbose) {
+                    String fmt = "%" + -groupLen + "s %" + -maxMemberIdLen + "s %" + -maxHostLen + "s %" + -maxClientIdLen + "s %s\n";
+                    System.out.printf(fmt, "GROUP", "MEMBER", "PROCESS", "CLIENT-ID", "ASSIGNMENTS");
+                    for (StreamsGroupMemberDescription member : members) {
+                        System.out.printf(fmt, description.groupId(), member.memberId(), member.processId(), member.clientId(),
+                            getTasksForPrinting(member.assignment(), Optional.empty()));
+                    }
+                } else {
+                    final int targetAssignmentEpochLen = 25, topologyEpochLen = 15, memberProtocolLen = 15, memberEpochLen = 15;
+                    String fmt = "%" + -groupLen + "s %" + -targetAssignmentEpochLen + "s %" + -topologyEpochLen + "s%" + -maxMemberIdLen
+                        + "s %" + -memberProtocolLen + "s %" + -memberEpochLen + "s %" + -maxHostLen + "s %" + -maxClientIdLen + "s %s\n";
+                    System.out.printf(fmt, "GROUP", "TARGET-ASSIGNMENT-EPOCH", "TOPOLOGY-EPOCH", "MEMBER", "MEMBER-PROTOCOL", "MEMBER-EPOCH", "PROCESS", "CLIENT-ID", "ASSIGNMENTS");
+                    for (StreamsGroupMemberDescription member : members) {
+                        System.out.printf(fmt, description.groupId(), description.targetAssignmentEpoch(), description.topologyEpoch(), member.memberId(),
+                            member.isClassic() ? "classic" : "streams", member.memberEpoch(), member.processId(), member.clientId(), getTasksForPrinting(member.assignment(), Optional.of(member.targetAssignment())));
+                    }
+                }
+            }
+        }
+
+        private String prepareTaskType(List<StreamsGroupMemberAssignment.TaskIds> tasks, String taskType) {
+            if (tasks.isEmpty()) {
+                return "";
+            }
+            StringBuilder builder = new StringBuilder(taskType).append(": ");
+            for (StreamsGroupMemberAssignment.TaskIds taskIds : tasks) {
+                builder.append(taskIds.subtopologyId()).append(":[");
+                builder.append(taskIds.partitions().stream().map(String::valueOf).collect(Collectors.joining(",")));
+                builder.append("]; ");
+            }
+            return builder.toString();
+        }
+
+        private String getTasksForPrinting(StreamsGroupMemberAssignment assignment, Optional<StreamsGroupMemberAssignment> targetAssignment) {
+            StringBuilder builder = new StringBuilder();
+            builder.append(prepareTaskType(assignment.activeTasks(), "ACTIVE"))
+                .append(prepareTaskType(assignment.standbyTasks(), "STANDBY"))
+                .append(prepareTaskType(assignment.warmupTasks(), "WARMUP"));
+            targetAssignment.ifPresent(target -> builder.append(prepareTaskType(target.activeTasks(), "TARGET-ACTIVE"))
+                .append(prepareTaskType(target.standbyTasks(), "TARGET-STANDBY"))
+                .append(prepareTaskType(target.warmupTasks(), "TARGET-WARMUP")));
+            return builder.toString();
+        }
+
+        private void printStates(StreamsGroupDescription description, boolean verbose) {
+            maybePrintEmptyGroupState(description.groupId(), description.groupState());
+
+            final int groupLen = Math.max(15, description.groupId().length());
+            String coordinator = description.coordinator().host() + ":" + description.coordinator().port() + " (" + description.coordinator().idString() + ")";
+
+            final int coordinatorLen = Math.max(25, coordinator.length());
+            final int stateLen = 25;
+            if (!verbose) {
+                String fmt = "%" + -groupLen + "s %" + -coordinatorLen + "s %" + -stateLen + "s %s\n";
+                System.out.printf(fmt, "GROUP", "COORDINATOR (ID)", "STATE", "#MEMBERS");
+                System.out.printf(fmt, description.groupId(), coordinator, description.groupState().toString(), description.members().size());
+            } else {
+                final int groupEpochLen = 15, targetAssignmentEpochLen = 25;
+                String fmt = "%" + -groupLen + "s %" + -coordinatorLen + "s %" + -stateLen + "s %" + -groupEpochLen + "s %" + -targetAssignmentEpochLen + "s %s\n";
+                System.out.printf(fmt, "GROUP", "COORDINATOR (ID)", "STATE", "GROUP-EPOCH", "TARGET-ASSIGNMENT-EPOCH", "#MEMBERS");
+                System.out.printf(fmt, description.groupId(), coordinator, description.groupState().toString(), description.groupEpoch(), description.targetAssignmentEpoch(), description.members().size());
+            }
+        }
+
+        private void printOffsets(StreamsGroupDescription description, boolean verbose) throws ExecutionException, InterruptedException {
+            Map<TopicPartition, OffsetsInfo> offsets = getOffsets(description);
+            if (isGroupStateValid(description.groupState(), description.members().size())) {
+                maybePrintEmptyGroupState(description.groupId(), description.groupState());
+                final int groupLen = Math.max(15, description.groupId().length());
+                int maxTopicLen = 15;
+                for (TopicPartition topicPartition : offsets.keySet()) {
+                    maxTopicLen = Math.max(maxTopicLen, topicPartition.topic().length());
+                }
+                final int maxPartitionLen = 10;
+                if (!verbose) {
+                    String fmt = "%" + -groupLen + "s %" + -maxTopicLen + "s %" + -maxPartitionLen + "s %s\n";
+                    System.out.printf(fmt, "GROUP", "TOPIC", "PARTITION", "OFFSET-LAG");
+                    for (Map.Entry<TopicPartition, OffsetsInfo> offset : offsets.entrySet()) {
+                        System.out.printf(fmt, description.groupId(), offset.getKey().topic(), offset.getKey().partition(), offset.getValue().lag);
+                    }
+                } else {
+                    String fmt = "%" + (-groupLen) + "s %" + (-maxTopicLen) + "s %-10s %-15s %-15s %-15s %-15s%n";
+                    System.out.printf(fmt, "GROUP", "TOPIC", "PARTITION", "CURRENT-OFFSET", "LEADER-EPOCH", "LOG-END-OFFSET", "OFFSET-LAG");
+                    for (Map.Entry<TopicPartition, OffsetsInfo> offset : offsets.entrySet()) {
+                        System.out.printf(fmt, description.groupId(), offset.getKey().topic(), offset.getKey().partition(),
+                            offset.getValue().currentOffset.map(Object::toString).orElse("-"), offset.getValue().leaderEpoch.map(Object::toString).orElse("-"),
+                            offset.getValue().logEndOffset, offset.getValue().lag);
+                    }
+                }
+            }
+        }
+
+        Map<TopicPartition, OffsetsInfo> getOffsets(StreamsGroupDescription description) throws ExecutionException, InterruptedException {
+            final Collection<StreamsGroupMemberDescription> members = description.members();
+            Set<TopicPartition> allTp = new HashSet<>();
+            for (StreamsGroupMemberDescription memberDescription : members) {
+                allTp.addAll(getTopicPartitions(memberDescription.assignment().activeTasks(), description));
+            }
+            // fetch latest and earliest offsets
+            Map<TopicPartition, OffsetSpec> earliest = new HashMap<>();
+            Map<TopicPartition, OffsetSpec> latest = new HashMap<>();
+
+            for (TopicPartition tp : allTp) {
+                earliest.put(tp, OffsetSpec.earliest());
+                latest.put(tp, OffsetSpec.latest());
+            }
+            Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> earliestResult = adminClient.listOffsets(earliest).all().get();
+            Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> latestResult = adminClient.listOffsets(latest).all().get();
+            Map<TopicPartition, OffsetAndMetadata> committedOffsets = getCommittedOffsets(description.groupId());
+
+            Map<TopicPartition, OffsetsInfo> output = new HashMap<>();
+            for (Map.Entry<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> tp : earliestResult.entrySet()) {
+                final Optional<Long> currentOffset = committedOffsets.containsKey(tp.getKey()) ? Optional.of(committedOffsets.get(tp.getKey()).offset()) : Optional.empty();
+                final Optional<Integer> leaderEpoch = committedOffsets.containsKey(tp.getKey()) ? committedOffsets.get(tp.getKey()).leaderEpoch() : Optional.empty();
+                final long lag = currentOffset.map(current -> latestResult.get(tp.getKey()).offset() - current).orElseGet(() -> latestResult.get(tp.getKey()).offset() - earliestResult.get(tp.getKey()).offset());
+                output.put(tp.getKey(),
+                    new OffsetsInfo(
+                        currentOffset,
+                        leaderEpoch,
+                        latestResult.get(tp.getKey()).offset(),
+                        lag));
+            }
+            return output;
+        }
+
+        Map<TopicPartition, OffsetAndMetadata> getCommittedOffsets(String groupId) {
+            try {
+                return adminClient.listConsumerGroupOffsets(
+                    Map.of(groupId, new ListConsumerGroupOffsetsSpec())).partitionsToOffsetAndMetadata(groupId).get();
+            } catch (InterruptedException | ExecutionException e) {
+                throw new RuntimeException(e);
+            }
+        }
+
+        /**
+         * Prints an error message if the group state indicates that the group is either dead or empty.
+         *
+         * @param group The ID of the group being checked.
+         * @param state The current state of the group, represented as a `GroupState` object.
+         *              Possible values include `DEAD` (indicating the group does not exist)
+         *              and `EMPTY` (indicating the group has no active members).
+         */
+        private static void maybePrintEmptyGroupState(String group, GroupState state) {
+            if (state == GroupState.DEAD) {
+                printError("Streams group '" + group + "' does not exist.", Optional.empty());
+            } else if (state == GroupState.EMPTY) {
+                printError("Streams group '" + group + "' has no active members.", Optional.empty());
+            }
+        }
+
+        /**
+         * Checks if the group state is valid based on its state and the number of rows.
+         *
+         * @param state   The current state of the group, represented as a `GroupState` object.
+         * @param numRows The number of rows associated with the group.
+         * @return `true` if the group state is not `DEAD` and the number of rows is greater than 0; otherwise, `false`.
+         */
+        // Visibility for testing
+        static boolean isGroupStateValid(GroupState state, int numRows) {
+            return !state.equals(GroupState.DEAD) && numRows > 0;
+        }
+
+        private static Set<TopicPartition> getTopicPartitions(List<StreamsGroupMemberAssignment.TaskIds> taskIds, StreamsGroupDescription description) {
+            Map<String, List<String>> allSourceTopics = new HashMap<>();
+            for (StreamsGroupSubtopologyDescription subtopologyDescription : description.subtopologies()) {
+                allSourceTopics.put(subtopologyDescription.subtopologyId(), subtopologyDescription.sourceTopics());
+            }
+            Set<TopicPartition> topicPartitions = new HashSet<>();
+
+            for (StreamsGroupMemberAssignment.TaskIds task : taskIds) {
+                List<String> sourceTopics = allSourceTopics.get(task.subtopologyId());
+                if (sourceTopics == null) {
+                    throw new IllegalArgumentException("Subtopology " + task.subtopologyId() + " not found in group description!");
+                }
+                for (String topic : sourceTopics) {
+                    for (Integer partition : task.partitions()) {
+                        topicPartitions.add(new TopicPartition(topic, partition));
+                    }
+                }
+            }
+            return topicPartitions;
+        }
+
         public void close() {
             adminClient.close();
         }
@@ -165,4 +397,7 @@ protected Admin createAdminClient(Map<String, String> configOverrides) throws IO
             return Admin.create(props);
         }
     }
+
+    public record OffsetsInfo(Optional<Long> currentOffset, Optional<Integer> leaderEpoch, Long logEndOffset, Long lag) {
+    }
 }