[GitHub] [kafka] dengziming commented on a diff in pull request #13562: KAFKA-14581: Moving GetOffsetShell to tools

["dengziming (via GitHub)" <gi...@apache.org>]

dengziming commented on code in PR #13562:
URL: https://github.com/apache/kafka/pull/13562#discussion_r1230929224


##########
tools/src/main/java/org/apache/kafka/tools/GetOffsetShell.java:
##########
@@ -0,0 +1,340 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.tools;
+
+import joptsimple.OptionParser;
+import joptsimple.OptionSet;
+import joptsimple.OptionSpec;
+import joptsimple.OptionSpecBuilder;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.ListOffsetsResult;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.ListTopicsOptions;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.requests.ListOffsetsRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.utils.Exit;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.util.CommandLineUtils;
+import org.apache.kafka.server.util.PartitionFilter;
+import org.apache.kafka.server.util.PartitionFilter.PartitionRangeFilter;
+import org.apache.kafka.server.util.PartitionFilter.PartitionsSetFilter;
+import org.apache.kafka.server.util.PartitionFilter.UniquePartitionFilter;
+import org.apache.kafka.server.util.TopicFilter.IncludeList;
+import org.apache.kafka.server.util.TopicPartitionFilter;
+import org.apache.kafka.server.util.TopicPartitionFilter.CompositeTopicPartitionFilter;
+import org.apache.kafka.server.util.TopicPartitionFilter.TopicFilterAndPartitionFilter;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.concurrent.ExecutionException;
+import java.util.function.IntFunction;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+public class GetOffsetShell {
+    Pattern topicPartitionPattern = Pattern.compile("([^:,]*)(?::(?:([0-9]*)|(?:([0-9]*)-([0-9]*))))?");
+
+    public static void main(String... args) {
+        Exit.exit(mainNoExit(args));
+    }
+
+    static int mainNoExit(String... args) {
+        try {
+            execute(args);
+            return 0;
+        } catch (Throwable e) {
+            System.err.println(e.getMessage());
+            System.err.println(Utils.stackTrace(e));
+            return 1;
+        }
+    }
+
+    static void execute(String... args) throws IOException, ExecutionException, InterruptedException {
+        GetOffsetShell getOffsetShell = new GetOffsetShell();
+
+        getOffsetShell.parseArgs(args);
+
+        Map<TopicPartition, Long> partitionOffsets = getOffsetShell.fetchOffsets();
+
+        for (Map.Entry<TopicPartition, Long> entry : partitionOffsets.entrySet()) {
+            TopicPartition topic = entry.getKey();
+
+            System.out.println(String.join(":", new String[]{topic.topic(), String.valueOf(topic.partition()), entry.getValue().toString()}));
+        }
+    }
+
+    private OptionSet options;
+    private OptionSpec<String> topicPartitionsOpt;
+    private OptionSpec<String> topicOpt;
+    private OptionSpec<String> partitionsOpt;
+    private OptionSpec<String> timeOpt;
+    private OptionSpec<String> commandConfigOpt;
+    private OptionSpec<String> effectiveBrokerListOpt;
+    private OptionSpecBuilder excludeInternalTopicsOpt;
+
+    public void parseArgs(final String[] args) {
+        final OptionParser parser = new OptionParser(false);
+
+        OptionSpec<String> brokerListOpt = parser.accepts("broker-list", "DEPRECATED, use --bootstrap-server instead; ignored if --bootstrap-server is specified. The server(s) to connect to in the form HOST1:PORT1,HOST2:PORT2.")
+                .withRequiredArg()
+                .describedAs("HOST1:PORT1,...,HOST3:PORT3")
+                .ofType(String.class);
+        OptionSpec<String> bootstrapServerOpt = parser.accepts("bootstrap-server", "REQUIRED. The server(s) to connect to in the form HOST1:PORT1,HOST2:PORT2.")
+                .requiredUnless("broker-list")
+                .withRequiredArg()
+                .describedAs("HOST1:PORT1,...,HOST3:PORT3")
+                .ofType(String.class);
+        topicPartitionsOpt = parser.accepts("topic-partitions", "Comma separated list of topic-partition patterns to get the offsets for, with the format of '" + topicPartitionPattern + "'." +
+                        " The first group is an optional regex for the topic name, if omitted, it matches any topic name." +
+                        " The section after ':' describes a 'partition' pattern, which can be: a number, a range in the format of 'NUMBER-NUMBER' (lower inclusive, upper exclusive), an inclusive lower bound in the format of 'NUMBER-', an exclusive upper bound in the format of '-NUMBER' or may be omitted to accept all partitions.")
+                .withRequiredArg()
+                .describedAs("topic1:1,topic2:0-3,topic3,topic4:5-,topic5:-3")
+                .ofType(String.class);
+        topicOpt = parser.accepts("topic", "The topic to get the offsets for. It also accepts a regular expression. If not present, all authorized topics are queried. Cannot be used if --topic-partitions is present.")
+                .withRequiredArg()
+                .describedAs("topic")
+                .ofType(String.class);
+        partitionsOpt = parser.accepts("partitions", "Comma separated list of partition ids to get the offsets for. If not present, all partitions of the authorized topics are queried. Cannot be used if --topic-partitions is present.")
+                .withRequiredArg()
+                .describedAs("partition ids")
+                .ofType(String.class);
+        timeOpt = parser.accepts("time", "timestamp of the offsets before that. [Note: No offset is returned, if the timestamp greater than recently committed record timestamp is given.]")
+                .withRequiredArg()
+                .describedAs("<timestamp> / -1 or latest / -2 or earliest / -3 or max-timestamp")
+                .ofType(String.class)
+                .defaultsTo("latest");
+        commandConfigOpt = parser.accepts("command-config", "Property file containing configs to be passed to Admin Client.")
+                .withRequiredArg()
+                .describedAs("config file")
+                .ofType(String.class);
+        excludeInternalTopicsOpt = parser.accepts("exclude-internal-topics", "By default, internal topics are included. If specified, internal topics are excluded.");
+
+        if (args.length == 0) {
+            CommandLineUtils.printUsageAndExit(parser, "An interactive shell for getting topic-partition offsets.");
+        }
+
+        options = parser.parse(args);
+
+        if (options.has(bootstrapServerOpt)) {
+            effectiveBrokerListOpt = bootstrapServerOpt;
+        } else {
+            effectiveBrokerListOpt = brokerListOpt;
+        }
+
+        CommandLineUtils.checkRequiredArgs(parser, options, effectiveBrokerListOpt);
+
+        String brokerList = options.valueOf(effectiveBrokerListOpt);
+
+        ToolsUtils.validatePortOrDie(parser, brokerList);
+    }
+
+    public Map<TopicPartition, Long> fetchOffsets() throws IOException, ExecutionException, InterruptedException {
+        String clientId = "GetOffsetShell";
+        String brokerList = options.valueOf(effectiveBrokerListOpt);
+
+        if (options.has(topicPartitionsOpt) && (options.has(topicOpt) || options.has(partitionsOpt))) {
+            throw new IllegalArgumentException("--topic-partitions cannot be used with --topic or --partitions");
+        }
+
+        boolean excludeInternalTopics = options.has(excludeInternalTopicsOpt);
+        OffsetSpec offsetSpec = parseOffsetSpec(options.valueOf(timeOpt));
+
+        TopicPartitionFilter topicPartitionFilter;

Review Comment:
   I didn't find this class, did you forgot to add it.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org