draft

Author: JimmyWang6

tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java

@@ -28,6 +28,7 @@
 import org.apache.kafka.clients.admin.ListGroupsOptions;
 import org.apache.kafka.clients.admin.ListGroupsResult;
 import org.apache.kafka.clients.admin.ListShareGroupOffsetsSpec;
+import org.apache.kafka.clients.admin.OffsetSpec;
 import org.apache.kafka.clients.admin.ShareGroupDescription;
 import org.apache.kafka.clients.admin.ShareMemberAssignment;
 import org.apache.kafka.clients.admin.ShareMemberDescription;
@@ -182,6 +183,10 @@ List<GroupListing> listShareGroupsInStates(Set<GroupState> states) throws Execut
             return new ArrayList<>(result.all().get());
         }
 
+        Map<TopicPartition, ConsumerGroupCommand.LogOffsetResult> getLogOffsets(Collection<TopicPartition> topicPartitions, OffsetSpec offsetSpec) {
+
+        }
+
         private void printGroupInfo(List<GroupListing> groups) {
             // find proper columns width
             int maxGroupLen = 15;
@@ -370,8 +375,7 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
 
         void resetOffsets() throws ExecutionException, InterruptedException {
             String groupId = opts.options.valueOf(opts.groupOpt);
-            Collection<SharePartitionOffsetInformation> partitionsToReset = getPartitionsToReset(groupId);
-            Map<TopicPartition, OffsetAndMetadata> offsetsToReset = getOffsetsToReset(partitionsToReset);
+            Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
             boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
             if (!dryRun) {
                 adminClient.alterConsumerGroupOffsets(groupId, offsetsToReset, new AlterConsumerGroupOffsetsOptions()
@@ -381,39 +385,41 @@ void resetOffsets() throws ExecutionException, InterruptedException {
             printOffsetFormat(groupId, partitionsToReset, opts.options.has(opts.verboseOpt));
         }
 
-        private Map<TopicPartition, OffsetAndMetadata> getOffsetsToReset(Collection<SharePartitionOffsetInformation> partitionsToReset) {
+        private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
             Map<TopicPartition, OffsetAndMetadata> offsetsToReset = new HashMap<>();
+            Map<String, ListShareGroupOffsetsSpec> groupSpecs = new HashMap<>();
+            Map<TopicPartition, OffsetAndMetadata> startOffsets = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+
+            if (opts.options.has(opts.topicOpt)) {
+                Set<String> topics = new HashSet<>(opts.options.valuesOf(opts.topicOpt));
+                startOffsets = startOffsets.entrySet().stream()
+                    .filter(entry -> {
+                        TopicPartition topicPartition = entry.getKey();
+                        return topics.contains(topicPartition.topic());
+                    })
+                    .collect(Collectors.toMap(
+                        Map.Entry::getKey,
+                        Map.Entry::getValue
+                    ));
+            }
+            if (startOffsets.isEmpty()) {
+                return Map.of();
+            }
+
             for (SharePartitionOffsetInformation partition : partitionsToReset) {
                 TopicPartition tp = new TopicPartition(partition.topic, partition.partition);
                 if (opts.options.has(opts.resetToEarliestOpt)) {
-                    offsetsToReset.put(tp, new OffsetAndMetadata(0L));
+                    //TODO implement resetToEarliestOpt
                 } else if (opts.options.has(opts.resetToLatestOpt)) {
-                    offsetsToReset.put(tp, new OffsetAndMetadata(Long.MAX_VALUE));
+                    //TODO implement resetToLatestOpt
                 } else if (opts.options.has(opts.resetToDatetimeOpt)) {
+                    //TODO implement resetToLatestOpt
                     offsetsToReset.put(tp, new OffsetAndMetadata(partition.offset.get()));
                 }
             }
             return offsetsToReset;
         }
 
-        private Collection<SharePartitionOffsetInformation> getPartitionsToReset(String groupId) throws ExecutionException, InterruptedException {
-           TreeMap<String, Entry<ShareGroupDescription, Collection<SharePartitionOffsetInformation>>> offsets
-               = collectGroupsOffsets(Set.of(groupId));
-           if (offsets.isEmpty()) {
-               return Collections.emptySet();
-           }
-           Collection<SharePartitionOffsetInformation> partitionOffsetInformation = offsets.get(groupId).getValue();
-           if (opts.options.has(opts.allTopicsOpt)) {
-               return partitionOffsetInformation;
-           } else if (opts.options.has(opts.topicOpt)) {
-               Set<String> topics= new HashSet<>(opts.options.valuesOf(opts.topicOpt));
-               return partitionOffsetInformation.stream()
-                   .filter(partition -> topics.contains(partition.topic))
-                   .collect(Collectors.toSet());
-           }
-           return Collections.emptySet();
-        }
-
         private <T extends AbstractOptions<T>> T withTimeoutMs(T options) {
             int t = opts.options.valueOf(opts.timeoutMsOpt).intValue();
             return options.timeoutMs(t);