From 7b989e430969273298329640a98ae3c0fd4de815 Mon Sep 17 00:00:00 2001
From: Daniel Urban <durban@cloudera.com>
Date: Mon, 29 Jun 2020 15:09:57 +0200
Subject: [PATCH 1/7] KAFKA-5235: GetOffsetShell: support for multiple topics
 and consumer configuration override

Implements KIP-635
---
 bin/kafka-get-offsets.sh                      |  17 ++
 .../scala/kafka/tools/GetOffsetShell.scala    | 198 ++++++++++++++----
 tests/kafkatest/services/kafka/kafka.py       |  14 +-
 .../tests/core/get_offset_shell_test.py       | 196 +++++++++++++++--
 4 files changed, 359 insertions(+), 66 deletions(-)
 create mode 100755 bin/kafka-get-offsets.sh

diff --git a/bin/kafka-get-offsets.sh b/bin/kafka-get-offsets.sh
new file mode 100755
index 0000000000000..993a202683309
--- /dev/null
+++ b/bin/kafka-get-offsets.sh
@@ -0,0 +1,17 @@
+#!/bin/bash
+# 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.
+
+exec $(dirname $0)/kafka-run-class.sh kafka.tools.GetOffsetShell "$@"
diff --git a/core/src/main/scala/kafka/tools/GetOffsetShell.scala b/core/src/main/scala/kafka/tools/GetOffsetShell.scala
index 7606247ebd6c1..0535d51f07a2f 100644
--- a/core/src/main/scala/kafka/tools/GetOffsetShell.scala
+++ b/core/src/main/scala/kafka/tools/GetOffsetShell.scala
@@ -20,11 +20,12 @@ package kafka.tools
 
 import java.util.Properties
 import joptsimple._
-import kafka.utils.{CommandLineUtils, Exit, ToolsUtils}
+import kafka.utils.{CommandLineUtils, Exit, ToolsUtils, IncludeList}
 import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
 import org.apache.kafka.common.requests.ListOffsetsRequest
 import org.apache.kafka.common.{PartitionInfo, TopicPartition}
 import org.apache.kafka.common.serialization.ByteArrayDeserializer
+import org.apache.kafka.common.utils.Utils
 
 import scala.jdk.CollectionConverters._
 import scala.collection.Seq
@@ -37,11 +38,17 @@ object GetOffsetShell {
                            .withRequiredArg
                            .describedAs("hostname:port,...,hostname:port")
                            .ofType(classOf[String])
-    val topicOpt = parser.accepts("topic", "REQUIRED: The topic to get offset from.")
+    val topicPartitionOpt = parser.accepts("topic-partitions", "Comma separated list of topic-partition specifications to get the offsets for, with the format of topic:partition. The 'topic' part can be a regex or may be omitted to only specify the partitions, and query all topics." +
+                                            " The 'partition' part 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 of the specified topic.")
+                           .withRequiredArg
+                           .describedAs("topic:partition,...,topic:partition")
+                           .ofType(classOf[String])
+                           .defaultsTo("")
+    val topicOpt = parser.accepts("topic", s"The topic to get the offsets for. It also accepts a regular expression. If not present, all topics are queried. Ignored if $topicPartitionOpt is present.")
                            .withRequiredArg
                            .describedAs("topic")
                            .ofType(classOf[String])
-    val partitionOpt = parser.accepts("partitions", "comma separated list of partition ids. If not specified, it will find offsets for all partitions")
+    val partitionOpt = parser.accepts("partitions", s"Comma separated list of partition ids to get the offsets for. If not present, all partitions of the topics are queried. Ignored if $topicPartitionOpt is present.")
                            .withRequiredArg
                            .describedAs("partition ids")
                            .ofType(classOf[String])
@@ -51,28 +58,24 @@ object GetOffsetShell {
                            .describedAs("timestamp/-1(latest)/-2(earliest)")
                            .ofType(classOf[java.lang.Long])
                            .defaultsTo(-1L)
-    parser.accepts("offsets", "DEPRECATED AND IGNORED: number of offsets returned")
+    val commandConfigOpt = parser.accepts("command-config", s"Consumer config properties file.")
                            .withRequiredArg
-                           .describedAs("count")
-                           .ofType(classOf[java.lang.Integer])
-                           .defaultsTo(1)
-    parser.accepts("max-wait-ms", "DEPRECATED AND IGNORED: The max amount of time each fetch request waits.")
-                           .withRequiredArg
-                           .describedAs("ms")
-                           .ofType(classOf[java.lang.Integer])
-                           .defaultsTo(1000)
+                           .describedAs("config file")
+                           .ofType(classOf[String])
+    val excludeInternalTopicsOpt = parser.accepts("exclude-internal-topics", s"By default, internal topics are included. If specified, internal topics are excluded.")
 
-   if (args.length == 0)
-      CommandLineUtils.printUsageAndDie(parser, "An interactive shell for getting topic offsets.")
+    if (args.length == 0)
+      CommandLineUtils.printUsageAndDie(parser, "An interactive shell for getting topic-partition offsets.")
 
     val options = parser.parse(args : _*)
 
-    CommandLineUtils.checkRequiredArgs(parser, options, brokerListOpt, topicOpt)
+    CommandLineUtils.checkRequiredArgs(parser, options, brokerListOpt)
 
     val clientId = "GetOffsetShell"
     val brokerList = options.valueOf(brokerListOpt)
     ToolsUtils.validatePortOrDie(parser, brokerList)
-    val topic = options.valueOf(topicOpt)
+    val excludeInternalTopics = options.has(excludeInternalTopicsOpt)
+
     val partitionIdsRequested: Set[Int] = {
       val partitionsString = options.valueOf(partitionOpt)
       if (partitionsString.isEmpty)
@@ -89,33 +92,40 @@ object GetOffsetShell {
     }
     val listOffsetsTimestamp = options.valueOf(timeOpt).longValue
 
-    val config = new Properties
+    val topicPartitionFilter = if (options.has(topicPartitionOpt)) {
+      Some(createTopicPartitionFilterWithPatternList(options.valueOf(topicPartitionOpt), excludeInternalTopics))
+    } else {
+      createTopicPartitionFilterWithTopicAndPartitionPattern(
+        if (options.has(topicOpt)) Some(options.valueOf(topicOpt)) else None,
+        excludeInternalTopics,
+        partitionIdsRequested
+      )
+    }
+
+    val config = if (options.has(commandConfigOpt))
+      Utils.loadProps(options.valueOf(commandConfigOpt))
+    else
+      new Properties
     config.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList)
     config.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, clientId)
     val consumer = new KafkaConsumer(config, new ByteArrayDeserializer, new ByteArrayDeserializer)
 
-    val partitionInfos = listPartitionInfos(consumer, topic, partitionIdsRequested) match {
-      case None =>
-        System.err.println(s"Topic $topic does not exist")
-        Exit.exit(1)
-      case Some(p) if p.isEmpty =>
-        if (partitionIdsRequested.isEmpty)
-          System.err.println(s"Topic $topic has 0 partitions")
-        else
-          System.err.println(s"Topic $topic does not have any of the requested partitions ${partitionIdsRequested.mkString(",")}")
-        Exit.exit(1)
-      case Some(p) => p
-    }
+    val partitionInfos = listPartitionInfos(consumer, topicPartitionFilter)
 
-    if (partitionIdsRequested.nonEmpty) {
-      (partitionIdsRequested -- partitionInfos.map(_.partition)).foreach { partitionId =>
-        System.err.println(s"Error: partition $partitionId does not exist")
-      }
+    if (partitionInfos.isEmpty) {
+      System.err.println(s"Could not match any topic-partitions with the specified filters")
+      Exit.exit(1)
     }
 
-    val topicPartitions = partitionInfos.sortBy(_.partition).flatMap { p =>
+    val topicPartitions = partitionInfos.sortWith((tp1, tp2) => {
+      val topicComp = tp1.topic.compareTo(tp2.topic)
+      if(topicComp == 0)
+        tp1.partition < tp2.partition
+      else
+        topicComp < 0
+    }).flatMap { p =>
       if (p.leader == null) {
-        System.err.println(s"Error: partition ${p.partition} does not have a leader. Skip getting offsets")
+        System.err.println(s"Error: topic-partition ${p.topic}:${p.partition} does not have a leader. Skip getting offsets")
         None
       } else
         Some(new TopicPartition(p.topic, p.partition))
@@ -132,23 +142,119 @@ object GetOffsetShell {
         }
     }
 
-    partitionOffsets.toArray.sortBy { case (tp, _) => tp.partition }.foreach { case (tp, offset) =>
-      println(s"$topic:${tp.partition}:${Option(offset).getOrElse("")}")
+    partitionOffsets.toSeq.sortWith((tp1, tp2) => {
+      val topicComp = tp1._1.topic.compareTo(tp2._1.topic)
+      if (topicComp == 0)
+        tp1._1.partition < tp2._1.partition
+      else
+        topicComp < 0
+    }).foreach { case (tp, offset) =>
+      println(s"${tp.topic}:${tp.partition}:${Option(offset).getOrElse("")}")
     }
 
   }
 
   /**
-   * Return the partition infos for `topic`. If the topic does not exist, `None` is returned.
+   * Creates a topic-partition filter based on a list of patterns.
+   * Expected format:
+   * List: TopicPartitionPattern(, TopicPartitionPattern)*
+   * TopicPartitionPattern: TopicPattern(:PartitionPattern)? | :PartitionPattern
+   * TopicPattern: REGEX
+   * PartitionPattern: NUMBER | NUMBER-(NUMBER)? | -NUMBER
    */
-  private def listPartitionInfos(consumer: KafkaConsumer[_, _], topic: String, partitionIds: Set[Int]): Option[Seq[PartitionInfo]] = {
-    val partitionInfos = consumer.listTopics.asScala.filter { case (k, _) => k == topic }.values.flatMap(_.asScala).toBuffer
-    if (partitionInfos.isEmpty)
-      None
-    else if (partitionIds.isEmpty)
-      Some(partitionInfos)
-    else
-      Some(partitionInfos.filter(p => partitionIds.contains(p.partition)))
+  private def createTopicPartitionFilterWithPatternList(topicPartitions: String, excludeInternalTopics: Boolean): PartitionInfo => Boolean = {
+    val ruleSpecs = topicPartitions.split(",")
+    val rules = ruleSpecs.map(ruleSpec => {
+      val parts = ruleSpec.split(":")
+      if (parts.length == 1) {
+        val whitelist = IncludeList(parts(0))
+        tp: PartitionInfo => whitelist.isTopicAllowed(tp.topic, excludeInternalTopics)
+      } else if (parts.length == 2) {
+        val partitionFilter = createPartitionFilter(parts(1))
+
+        if (parts(0).trim().isEmpty) {
+          tp: PartitionInfo => partitionFilter.apply(tp.partition)
+        } else {
+          val whitelist = IncludeList(parts(0))
+          tp: PartitionInfo => whitelist.isTopicAllowed(tp.topic, excludeInternalTopics) && partitionFilter.apply(tp.partition)
+        }
+      } else {
+        throw new IllegalArgumentException(s"Invalid topic-partition rule: $ruleSpec")
+      }
+    })
+
+    tp => rules.exists(rule => rule.apply(tp))
   }
 
+  /**
+   * Creates a partition filter based on a single id or a range.
+   * Expected format:
+   * PartitionPattern: NUMBER | NUMBER-(NUMBER)? | -NUMBER
+   */
+  private def createPartitionFilter(spec: String): Int => Boolean = {
+    if (spec.indexOf('-') != -1) {
+      val rangeParts = spec.split("-", -1)
+      if(rangeParts.length != 2 || rangeParts(0).isEmpty && rangeParts(1).isEmpty) {
+        throw new IllegalArgumentException(s"Invalid range specification: $spec")
+      }
+
+      if(rangeParts(0).isEmpty) {
+        val max = rangeParts(1).toInt
+        partition: Int => partition < max
+      } else if(rangeParts(1).isEmpty) {
+        val min = rangeParts(0).toInt
+        partition: Int => partition >= min
+      } else {
+        val min = rangeParts(0).toInt
+        val max = rangeParts(1).toInt
+
+        if (min > max) {
+          throw new IllegalArgumentException(s"Range lower bound cannot be greater than upper bound: $spec")
+        }
+
+        partition: Int => partition >= min && partition < max
+      }
+    } else {
+      val number = spec.toInt
+      partition: Int => partition == number
+    }
+  }
+
+  /**
+   * Creates a topic-partition filter based on a topic pattern and a set of partition ids.
+   */
+  private def createTopicPartitionFilterWithTopicAndPartitionPattern(topicOpt: Option[String], excludeInternalTopics: Boolean, partitionIds: Set[Int]): Option[PartitionInfo => Boolean] = {
+    topicOpt match {
+      case Some(topic) =>
+        val topicsFilter = IncludeList(topic)
+        if(partitionIds.isEmpty)
+          Some(t => topicsFilter.isTopicAllowed(t.topic, excludeInternalTopics))
+        else
+          Some(t => topicsFilter.isTopicAllowed(t.topic, excludeInternalTopics) && partitionIds.contains(t.partition))
+      case None =>
+        if(excludeInternalTopics) {
+          if(partitionIds.isEmpty)
+            Some(t => !Topic.isInternal(t.topic))
+          else
+            Some(t => !Topic.isInternal(t.topic) && partitionIds.contains(t.partition))
+        } else {
+          if(partitionIds.isEmpty)
+            None
+          else
+            Some(t => partitionIds.contains(t.partition))
+        }
+    }
+  }
+
+  /**
+   * Return the partition infos. Filter them with topicPartitionFilter if specified.
+   */
+  private def listPartitionInfos(consumer: KafkaConsumer[_, _], topicPartitionFilter: Option[PartitionInfo => Boolean]): Seq[PartitionInfo] = {
+    val topicListUnfiltered = consumer.listTopics.asScala.values.flatMap { tp => tp.asScala }
+    val topicList = topicPartitionFilter match {
+      case Some(filter) => topicListUnfiltered.filter { tp => filter.apply(tp) }
+      case _ => topicListUnfiltered
+    }
+    topicList.toBuffer
+  }
 }
diff --git a/tests/kafkatest/services/kafka/kafka.py b/tests/kafkatest/services/kafka/kafka.py
index af5a753f7e51c..bb0dc691e8295 100644
--- a/tests/kafkatest/services/kafka/kafka.py
+++ b/tests/kafkatest/services/kafka/kafka.py
@@ -1152,16 +1152,24 @@ def is_registered(self, node):
         self.logger.debug("Broker info: %s", broker_info)
         return broker_info is not None
 
-    def get_offset_shell(self, topic, partitions, max_wait_ms, offsets, time):
+    def get_offset_shell(self, time=None, topic=None, partitions=None, topic_partitions=None, exclude_internal_topics=False):
         node = self.nodes[0]
 
         cmd = fix_opts_for_new_jvm(node)
         cmd += self.path.script("kafka-run-class.sh", node)
         cmd += " kafka.tools.GetOffsetShell"
-        cmd += " --topic %s --broker-list %s --max-wait-ms %s --offsets %s --time %s" % (topic, self.bootstrap_servers(self.security_protocol), max_wait_ms, offsets, time)
-
+        cmd += " --broker-list %s" % self.bootstrap_servers(self.security_protocol)
+
+        if time:
+            cmd += ' --time %s' % time
+        if topic_partitions:
+            cmd += ' --topic-partitions %s' % topic_partitions
+        if topic:
+            cmd += ' --topic %s' % topic
         if partitions:
             cmd += '  --partitions %s' % partitions
+        if exclude_internal_topics:
+            cmd += ' --exclude-internal-topics'
 
         cmd += " 2>> %s/get_offset_shell.log" % KafkaService.PERSISTENT_ROOT
         cmd += " | tee -a %s/get_offset_shell.log &" % KafkaService.PERSISTENT_ROOT
diff --git a/tests/kafkatest/tests/core/get_offset_shell_test.py b/tests/kafkatest/tests/core/get_offset_shell_test.py
index 82291d7db48c6..3f226c1d0bcc5 100644
--- a/tests/kafkatest/tests/core/get_offset_shell_test.py
+++ b/tests/kafkatest/tests/core/get_offset_shell_test.py
@@ -23,11 +23,26 @@
 from kafkatest.services.kafka import KafkaService
 from kafkatest.services.console_consumer import ConsoleConsumer
 
-TOPIC = "topic-get-offset-shell"
 MAX_MESSAGES = 100
 NUM_PARTITIONS = 1
 REPLICATION_FACTOR = 1
 
+TOPIC_TEST_NAME = "topic-get-offset-shell-topic-name"
+
+TOPIC_TEST_PATTERN_PREFIX = "topic-get-offset-shell-topic-pattern"
+TOPIC_TEST_PATTERN_PATTERN = TOPIC_TEST_PATTERN_PREFIX + ".*"
+TOPIC_TEST_PATTERN1 = TOPIC_TEST_PATTERN_PREFIX + "1"
+TOPIC_TEST_PATTERN2 = TOPIC_TEST_PATTERN_PREFIX + "2"
+
+TOPIC_TEST_PARTITIONS = "topic-get-offset-shell-partitions"
+
+TOPIC_TEST_INTERNAL_FILTER = "topic-get-offset-shell-consumer_offsets"
+
+TOPIC_TEST_TOPIC_PARTITIONS_PREFIX = "topic-get-offset-shell-topic-partitions"
+TOPIC_TEST_TOPIC_PARTITIONS_PATTERN = TOPIC_TEST_TOPIC_PARTITIONS_PREFIX + ".*"
+TOPIC_TEST_TOPIC_PARTITIONS1 = TOPIC_TEST_TOPIC_PARTITIONS_PREFIX + "1"
+TOPIC_TEST_TOPIC_PARTITIONS2 = TOPIC_TEST_TOPIC_PARTITIONS_PREFIX + "2"
+
 
 class GetOffsetShellTest(Test):
     """
@@ -39,12 +54,17 @@ def __init__(self, test_context):
         self.num_brokers = 1
         self.messages_received_count = 0
         self.topics = {
-            TOPIC: {'partitions': NUM_PARTITIONS, 'replication-factor': REPLICATION_FACTOR}
+            TOPIC_TEST_NAME: {'partitions': NUM_PARTITIONS, 'replication-factor': REPLICATION_FACTOR},
+            TOPIC_TEST_PATTERN1: {'partitions': 1, 'replication-factor': REPLICATION_FACTOR},
+            TOPIC_TEST_PATTERN2: {'partitions': 1, 'replication-factor': REPLICATION_FACTOR},
+            TOPIC_TEST_PARTITIONS: {'partitions': 2, 'replication-factor': REPLICATION_FACTOR},
+            TOPIC_TEST_INTERNAL_FILTER: {'partitions': 1, 'replication-factor': REPLICATION_FACTOR},
+            TOPIC_TEST_TOPIC_PARTITIONS1: {'partitions': 2, 'replication-factor': REPLICATION_FACTOR},
+            TOPIC_TEST_TOPIC_PARTITIONS2: {'partitions': 2, 'replication-factor': REPLICATION_FACTOR}
         }
 
         self.zk = ZookeeperService(test_context, self.num_zk)
 
-
     def setUp(self):
         self.zk.start()
 
@@ -55,37 +75,179 @@ def start_kafka(self, security_protocol, interbroker_security_protocol):
             interbroker_security_protocol=interbroker_security_protocol, topics=self.topics)
         self.kafka.start()
 
-    def start_producer(self):
+    def start_producer(self, topic):
         # This will produce to kafka cluster
-        self.producer = VerifiableProducer(self.test_context, num_nodes=1, kafka=self.kafka, topic=TOPIC, throughput=1000, max_messages=MAX_MESSAGES)
+        self.producer = VerifiableProducer(self.test_context, num_nodes=1, kafka=self.kafka, topic=topic,
+                                           throughput=1000, max_messages=MAX_MESSAGES, repeating_keys=MAX_MESSAGES)
         self.producer.start()
         current_acked = self.producer.num_acked
         wait_until(lambda: self.producer.num_acked >= current_acked + MAX_MESSAGES, timeout_sec=10,
                    err_msg="Timeout awaiting messages to be produced and acked")
 
-    def start_consumer(self):
-        self.consumer = ConsoleConsumer(self.test_context, num_nodes=self.num_brokers, kafka=self.kafka, topic=TOPIC,
+    def start_consumer(self, topic):
+        self.consumer = ConsoleConsumer(self.test_context, num_nodes=self.num_brokers, kafka=self.kafka, topic=topic,
                                         consumer_timeout_ms=1000)
         self.consumer.start()
 
+    def check_message_count_sum_equals(self, message_count, **kwargs):
+        sum = self.extract_message_count_sum(**kwargs)
+        return sum == message_count
+
+    def extract_message_count_sum(self, **kwargs):
+        offsets = self.kafka.get_offset_shell(**kwargs).split("\n")
+        sum = 0
+        for offset in offsets:
+            if len(offset) == 0:
+                continue
+            sum += int(offset.split(":")[-1])
+        return sum
+
+    @cluster(num_nodes=3)
+    def test_get_offset_shell_topic_name(self, security_protocol='PLAINTEXT'):
+        """
+        Tests if GetOffsetShell handles --topic argument with a simple name correctly
+        :return: None
+        """
+        self.start_kafka(security_protocol, security_protocol)
+        self.start_producer(TOPIC_TEST_NAME)
+
+        # Assert that offset is correctly indicated by GetOffsetShell tool
+        wait_until(lambda: self.check_message_count_sum_equals(MAX_MESSAGES, topic=TOPIC_TEST_NAME),
+                   timeout_sec=10, err_msg="Timed out waiting to reach expected offset.")
+
     @cluster(num_nodes=4)
-    def test_get_offset_shell(self, security_protocol='PLAINTEXT'):
+    def test_get_offset_shell_topic_pattern(self, security_protocol='PLAINTEXT'):
         """
-        Tests if GetOffsetShell is getting offsets correctly
+        Tests if GetOffsetShell handles --topic argument with a pattern correctly
         :return: None
         """
         self.start_kafka(security_protocol, security_protocol)
-        self.start_producer()
+        self.start_producer(TOPIC_TEST_PATTERN1)
+        self.start_producer(TOPIC_TEST_PATTERN2)
 
-        # Assert that offset fetched without any consumers consuming is 0
-        assert self.kafka.get_offset_shell(TOPIC, None, 1000, 1, -1), "%s:%s:%s" % (TOPIC, NUM_PARTITIONS - 1, 0)
+        # Assert that offset is correctly indicated by GetOffsetShell tool
+        wait_until(lambda: self.check_message_count_sum_equals(2*MAX_MESSAGES, topic=TOPIC_TEST_PATTERN_PATTERN),
+                   timeout_sec=10, err_msg="Timed out waiting to reach expected offset.")
 
-        self.start_consumer()
+    @cluster(num_nodes=3)
+    def test_get_offset_shell_partitions(self, security_protocol='PLAINTEXT'):
+        """
+        Tests if GetOffsetShell handles --partitions argument correctly
+        :return: None
+        """
+        self.start_kafka(security_protocol, security_protocol)
+        self.start_producer(TOPIC_TEST_PARTITIONS)
 
-        node = self.consumer.nodes[0]
+        def fetch_and_sum_partitions_separately():
+            partition_count0 = self.extract_message_count_sum(topic=TOPIC_TEST_PARTITIONS, partitions="0")
+            partition_count1 = self.extract_message_count_sum(topic=TOPIC_TEST_PARTITIONS, partitions="1")
+            return partition_count0 + partition_count1 == MAX_MESSAGES
+
+        # Assert that offset is correctly indicated when fetching partitions one by one
+        wait_until(fetch_and_sum_partitions_separately, timeout_sec=10, err_msg="Timed out waiting to reach expected offset.")
+
+        # Assert that offset is correctly indicated when fetching partitions together
+        wait_until(lambda: self.check_message_count_sum_equals(MAX_MESSAGES, topic=TOPIC_TEST_PARTITIONS),
+                   timeout_sec=10, err_msg="Timed out waiting to reach expected offset.")
+
+    @cluster(num_nodes=4)
+    def test_get_offset_shell_topic_partitions(self, security_protocol='PLAINTEXT'):
+        """
+        Tests if GetOffsetShell handles --topic-partitions argument correctly
+        :return: None
+        """
+        self.start_kafka(security_protocol, security_protocol)
+        self.start_producer(TOPIC_TEST_TOPIC_PARTITIONS1)
+        self.start_producer(TOPIC_TEST_TOPIC_PARTITIONS2)
+
+        # Assert that a single topic pattern matches all 4 partitions
+        wait_until(lambda: self.check_message_count_sum_equals(2*MAX_MESSAGES, topic_partitions=TOPIC_TEST_TOPIC_PARTITIONS_PATTERN),
+                   timeout_sec=10, err_msg="Timed out waiting to reach expected offset.")
+
+        # Assert that a topic pattern with partition range matches all 4 partitions
+        wait_until(lambda: self.check_message_count_sum_equals(2*MAX_MESSAGES, topic_partitions=TOPIC_TEST_TOPIC_PARTITIONS_PATTERN + ":0-2"),
+                   timeout_sec=10, err_msg="Timed out waiting to reach expected offset.")
+
+        # Assert that 2 separate topic patterns match all 4 partitions
+        wait_until(lambda: self.check_message_count_sum_equals(2*MAX_MESSAGES, topic_partitions=TOPIC_TEST_TOPIC_PARTITIONS1 + "," + TOPIC_TEST_TOPIC_PARTITIONS2),
+                   timeout_sec=10, err_msg="Timed out waiting to reach expected offset.")
+
+        # Assert that 4 separate topic-partition patterns match all 4 partitions
+        wait_until(lambda: self.check_message_count_sum_equals(2*MAX_MESSAGES,
+                                                               topic_partitions=TOPIC_TEST_TOPIC_PARTITIONS1 + ":0," +
+                                                                                TOPIC_TEST_TOPIC_PARTITIONS1 + ":1," +
+                                                                                TOPIC_TEST_TOPIC_PARTITIONS2 + ":0," +
+                                                                                TOPIC_TEST_TOPIC_PARTITIONS2 + ":1"),
+                   timeout_sec=10, err_msg="Timed out waiting to reach expected offset.")
+
+        # Assert that only partitions #0 are matched with topic pattern and fix partition number
+        filtered_partitions = self.kafka.get_offset_shell(topic_partitions=TOPIC_TEST_TOPIC_PARTITIONS_PATTERN + ":0")
+        assert 1 == filtered_partitions.count("%s:%s" % (TOPIC_TEST_TOPIC_PARTITIONS1, 0))
+        assert 0 == filtered_partitions.count("%s:%s" % (TOPIC_TEST_TOPIC_PARTITIONS1, 1))
+        assert 1 == filtered_partitions.count("%s:%s" % (TOPIC_TEST_TOPIC_PARTITIONS2, 0))
+        assert 0 == filtered_partitions.count("%s:%s" % (TOPIC_TEST_TOPIC_PARTITIONS2, 1))
+
+        # Assert that only partitions #1 are matched with topic pattern and partition lower bound
+        filtered_partitions = self.kafka.get_offset_shell(topic_partitions=TOPIC_TEST_TOPIC_PARTITIONS_PATTERN + ":1-")
+        assert 1 == filtered_partitions.count("%s:%s" % (TOPIC_TEST_TOPIC_PARTITIONS1, 1))
+        assert 0 == filtered_partitions.count("%s:%s" % (TOPIC_TEST_TOPIC_PARTITIONS1, 0))
+        assert 1 == filtered_partitions.count("%s:%s" % (TOPIC_TEST_TOPIC_PARTITIONS2, 1))
+        assert 0 == filtered_partitions.count("%s:%s" % (TOPIC_TEST_TOPIC_PARTITIONS2, 0))
+
+        # Assert that only partitions #0 are matched with topic pattern and partition upper bound
+        filtered_partitions = self.kafka.get_offset_shell(topic_partitions=TOPIC_TEST_TOPIC_PARTITIONS_PATTERN + ":-1")
+        assert 1 == filtered_partitions.count("%s:%s" % (TOPIC_TEST_TOPIC_PARTITIONS1, 0))
+        assert 0 == filtered_partitions.count("%s:%s" % (TOPIC_TEST_TOPIC_PARTITIONS1, 1))
+        assert 1 == filtered_partitions.count("%s:%s" % (TOPIC_TEST_TOPIC_PARTITIONS2, 0))
+        assert 0 == filtered_partitions.count("%s:%s" % (TOPIC_TEST_TOPIC_PARTITIONS2, 1))
 
+    @cluster(num_nodes=4)
+    def test_get_offset_shell_internal_filter(self, security_protocol='PLAINTEXT'):
+        """
+        Tests if GetOffsetShell handles --exclude-internal-topics flag correctly
+        :return: None
+        """
+        self.start_kafka(security_protocol, security_protocol)
+        self.start_producer(TOPIC_TEST_INTERNAL_FILTER)
+
+        # Create consumer and poll messages to create consumer offset record
+        self.start_consumer(TOPIC_TEST_INTERNAL_FILTER)
+        node = self.consumer.nodes[0]
         wait_until(lambda: self.consumer.alive(node), timeout_sec=20, backoff_sec=.2, err_msg="Consumer was too slow to start")
 
-        # Assert that offset is correctly indicated by GetOffsetShell tool
-        wait_until(lambda: "%s:%s:%s" % (TOPIC, NUM_PARTITIONS - 1, MAX_MESSAGES) in self.kafka.get_offset_shell(TOPIC, None, 1000, 1, -1), timeout_sec=10,
-                   err_msg="Timed out waiting to reach expected offset.")
+        # Assert that a single topic pattern matches all 4 partitions
+        wait_until(lambda: self.check_message_count_sum_equals(MAX_MESSAGES, topic_partitions=TOPIC_TEST_INTERNAL_FILTER),
+                   timeout_sec=10, err_msg="Timed out waiting to reach expected offset.")
+
+        # No filters
+        # Assert that without exclusion, we can find both the test topic and the __consumer_offsets internal topic
+        offset_output = self.kafka.get_offset_shell()
+        assert "__consumer_offsets" in offset_output
+        assert TOPIC_TEST_INTERNAL_FILTER in offset_output
+
+        # Assert that with exclusion, we can find the test topic but not the __consumer_offsets internal topic
+        offset_output = self.kafka.get_offset_shell(exclude_internal_topics=True)
+        assert "__consumer_offsets" not in offset_output
+        assert TOPIC_TEST_INTERNAL_FILTER in offset_output
+
+        # Topic filter
+        # Assert that without exclusion, we can find both the test topic and the __consumer_offsets internal topic
+        offset_output = self.kafka.get_offset_shell(topic=".*consumer_offsets")
+        assert "__consumer_offsets" in offset_output
+        assert TOPIC_TEST_INTERNAL_FILTER in offset_output
+
+        # Assert that with exclusion, we can find the test topic but not the __consumer_offsets internal topic
+        offset_output = self.kafka.get_offset_shell(topic=".*consumer_offsets", exclude_internal_topics=True)
+        assert "__consumer_offsets" not in offset_output
+        assert TOPIC_TEST_INTERNAL_FILTER in offset_output
+
+        # Topic-partition filter
+        # Assert that without exclusion, we can find both the test topic and the __consumer_offsets internal topic
+        offset_output = self.kafka.get_offset_shell(topic_partitions=".*consumer_offsets:0")
+        assert "__consumer_offsets" in offset_output
+        assert TOPIC_TEST_INTERNAL_FILTER in offset_output
+
+        # Assert that with exclusion, we can find the test topic but not the __consumer_offsets internal topic
+        offset_output = self.kafka.get_offset_shell(topic_partitions=".*consumer_offsets:0", exclude_internal_topics=True)
+        assert "__consumer_offsets" not in offset_output
+        assert TOPIC_TEST_INTERNAL_FILTER in offset_output

From 0ae1a7580cea131bcccbd939a058a05451e073ae Mon Sep 17 00:00:00 2001
From: Daniel Urban <durban@cloudera.com>
Date: Fri, 17 Jul 2020 10:53:16 +0200
Subject: [PATCH 2/7] KAFKA-5235: Added .bat for windows, usage now mentions
 that only authorized topics are scanned

---
 bin/windows/kafka-get-offsets.bat               | 17 +++++++++++++++++
 .../main/scala/kafka/tools/GetOffsetShell.scala |  6 +++---
 2 files changed, 20 insertions(+), 3 deletions(-)
 create mode 100644 bin/windows/kafka-get-offsets.bat

diff --git a/bin/windows/kafka-get-offsets.bat b/bin/windows/kafka-get-offsets.bat
new file mode 100644
index 0000000000000..08b8e27d70fec
--- /dev/null
+++ b/bin/windows/kafka-get-offsets.bat
@@ -0,0 +1,17 @@
+@echo off
+rem Licensed to the Apache Software Foundation (ASF) under one or more
+rem contributor license agreements.  See the NOTICE file distributed with
+rem this work for additional information regarding copyright ownership.
+rem The ASF licenses this file to You under the Apache License, Version 2.0
+rem (the "License"); you may not use this file except in compliance with
+rem the License.  You may obtain a copy of the License at
+rem
+rem     http://www.apache.org/licenses/LICENSE-2.0
+rem
+rem Unless required by applicable law or agreed to in writing, software
+rem distributed under the License is distributed on an "AS IS" BASIS,
+rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+rem See the License for the specific language governing permissions and
+rem limitations under the License.
+
+"%~dp0kafka-run-class.bat" kafka.tools.GetOffsetShell %*
diff --git a/core/src/main/scala/kafka/tools/GetOffsetShell.scala b/core/src/main/scala/kafka/tools/GetOffsetShell.scala
index 0535d51f07a2f..019c1d82dd5da 100644
--- a/core/src/main/scala/kafka/tools/GetOffsetShell.scala
+++ b/core/src/main/scala/kafka/tools/GetOffsetShell.scala
@@ -38,17 +38,17 @@ object GetOffsetShell {
                            .withRequiredArg
                            .describedAs("hostname:port,...,hostname:port")
                            .ofType(classOf[String])
-    val topicPartitionOpt = parser.accepts("topic-partitions", "Comma separated list of topic-partition specifications to get the offsets for, with the format of topic:partition. The 'topic' part can be a regex or may be omitted to only specify the partitions, and query all topics." +
+    val topicPartitionOpt = parser.accepts("topic-partitions", "Comma separated list of topic-partition specifications to get the offsets for, with the format of topic:partition. The 'topic' part can be a regex or may be omitted to only specify the partitions, and query all authorized topics." +
                                             " The 'partition' part 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 of the specified topic.")
                            .withRequiredArg
                            .describedAs("topic:partition,...,topic:partition")
                            .ofType(classOf[String])
                            .defaultsTo("")
-    val topicOpt = parser.accepts("topic", s"The topic to get the offsets for. It also accepts a regular expression. If not present, all topics are queried. Ignored if $topicPartitionOpt is present.")
+    val topicOpt = parser.accepts("topic", s"The topic to get the offsets for. It also accepts a regular expression. If not present, all authorized topics are queried. Ignored if $topicPartitionOpt is present.")
                            .withRequiredArg
                            .describedAs("topic")
                            .ofType(classOf[String])
-    val partitionOpt = parser.accepts("partitions", s"Comma separated list of partition ids to get the offsets for. If not present, all partitions of the topics are queried. Ignored if $topicPartitionOpt is present.")
+    val partitionOpt = parser.accepts("partitions", s"Comma separated list of partition ids to get the offsets for. If not present, all partitions of the authorized topics are queried. Ignored if $topicPartitionOpt is present.")
                            .withRequiredArg
                            .describedAs("partition ids")
                            .ofType(classOf[String])

From fed78f70588afc06bbb77396fa77f237258828b2 Mon Sep 17 00:00:00 2001
From: Daniel Urban <durban@cloudera.com>
Date: Fri, 14 Aug 2020 14:24:33 +0200
Subject: [PATCH 3/7] KAFKA-5235: Deprecated --broker-list argument, introduced
 --boostrap-servre

---
 .../scala/kafka/tools/GetOffsetShell.scala    | 19 +++++++++++++++----
 tests/kafkatest/services/kafka/kafka.py       |  2 +-
 2 files changed, 16 insertions(+), 5 deletions(-)

diff --git a/core/src/main/scala/kafka/tools/GetOffsetShell.scala b/core/src/main/scala/kafka/tools/GetOffsetShell.scala
index 019c1d82dd5da..54224617fb312 100644
--- a/core/src/main/scala/kafka/tools/GetOffsetShell.scala
+++ b/core/src/main/scala/kafka/tools/GetOffsetShell.scala
@@ -34,9 +34,14 @@ object GetOffsetShell {
 
   def main(args: Array[String]): Unit = {
     val parser = new OptionParser(false)
-    val brokerListOpt = parser.accepts("broker-list", "REQUIRED: The list of hostname and port of the server to connect to.")
+    val 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("hostname:port,...,hostname:port")
+                           .describedAs("HOST1:PORT1,...,HOST3:PORT3")
+                           .ofType(classOf[String])
+    val 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(classOf[String])
     val topicPartitionOpt = parser.accepts("topic-partitions", "Comma separated list of topic-partition specifications to get the offsets for, with the format of topic:partition. The 'topic' part can be a regex or may be omitted to only specify the partitions, and query all authorized topics." +
                                             " The 'partition' part 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 of the specified topic.")
@@ -69,10 +74,16 @@ object GetOffsetShell {
 
     val options = parser.parse(args : _*)
 
-    CommandLineUtils.checkRequiredArgs(parser, options, brokerListOpt)
+    val effectiveBrokerListOpt = if (options.has(bootstrapServerOpt))
+      bootstrapServerOpt
+    else
+      brokerListOpt
+
+    CommandLineUtils.checkRequiredArgs(parser, options, effectiveBrokerListOpt)
 
     val clientId = "GetOffsetShell"
-    val brokerList = options.valueOf(brokerListOpt)
+    val brokerList = options.valueOf(effectiveBrokerListOpt)
+
     ToolsUtils.validatePortOrDie(parser, brokerList)
     val excludeInternalTopics = options.has(excludeInternalTopicsOpt)
 
diff --git a/tests/kafkatest/services/kafka/kafka.py b/tests/kafkatest/services/kafka/kafka.py
index bb0dc691e8295..40592f67be79d 100644
--- a/tests/kafkatest/services/kafka/kafka.py
+++ b/tests/kafkatest/services/kafka/kafka.py
@@ -1158,7 +1158,7 @@ def get_offset_shell(self, time=None, topic=None, partitions=None, topic_partiti
         cmd = fix_opts_for_new_jvm(node)
         cmd += self.path.script("kafka-run-class.sh", node)
         cmd += " kafka.tools.GetOffsetShell"
-        cmd += " --broker-list %s" % self.bootstrap_servers(self.security_protocol)
+        cmd += " --boostrap-server %s" % self.bootstrap_servers(self.security_protocol)
 
         if time:
             cmd += ' --time %s' % time

From e7d2f2b3ab1c41602731185807b919789e1cb2df Mon Sep 17 00:00:00 2001
From: Daniel Urban <durban@cloudera.com>
Date: Tue, 10 Nov 2020 11:57:35 +0100
Subject: [PATCH 4/7] KAFKA-5235: Unit tests for GetOffsetShell

---
 .../kafka/tools/GetOffsetShellTest.scala      | 194 ++++++++++++++++++
 1 file changed, 194 insertions(+)
 create mode 100644 core/src/test/scala/kafka/tools/GetOffsetShellTest.scala

diff --git a/core/src/test/scala/kafka/tools/GetOffsetShellTest.scala b/core/src/test/scala/kafka/tools/GetOffsetShellTest.scala
new file mode 100644
index 0000000000000..7b53733f605ed
--- /dev/null
+++ b/core/src/test/scala/kafka/tools/GetOffsetShellTest.scala
@@ -0,0 +1,194 @@
+package kafka.tools
+
+import java.time.Duration
+import java.util.Properties
+import java.util.regex.Pattern
+
+import kafka.integration.KafkaServerTestHarness
+import kafka.server.KafkaConfig
+import kafka.utils.{Exit, Logging, TestUtils}
+import org.apache.kafka.clients.CommonClientConfigs
+import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
+import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
+import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}
+import org.junit.Assert.{assertEquals, assertTrue}
+import org.junit.{Before, Test}
+
+class GetOffsetShellTest extends KafkaServerTestHarness with Logging {
+  private val topicCount = 4
+  private val offsetTopicPartitionCount = 4
+  private val topicPattern = Pattern.compile("test.*")
+
+  override def generateConfigs: collection.Seq[KafkaConfig] = TestUtils.createBrokerConfigs(1, zkConnect)
+    .map(p => {
+      p.put(KafkaConfig.OffsetsTopicPartitionsProp, offsetTopicPartitionCount)
+      p
+    }).map(KafkaConfig.fromProps)
+
+  @Before
+  def createTopicAndConsume(): Unit = {
+    Range(1, topicCount + 1).foreach(i => createTopic(topicName(i), i))
+
+    val props = new Properties()
+    props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerList)
+    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
+    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
+    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer])
+    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer])
+    props.put(ConsumerConfig.GROUP_ID_CONFIG, "GetOffsetShellTest")
+
+    // Send X messages to each partition of topicX
+    val producer = new KafkaProducer[String, String](props)
+    Range(1, topicCount + 1).foreach(i => Range(0, i*i)
+      .foreach(msgCount => producer.send(new ProducerRecord[String, String](topicName(i), msgCount % i, null, "val" + msgCount))))
+    producer.close()
+
+    // Consume so consumer offsets topic is created
+    val consumer = new KafkaConsumer[String, String](props)
+    consumer.subscribe(topicPattern)
+    consumer.poll(Duration.ofMillis(1000))
+    consumer.commitSync()
+    consumer.close()
+  }
+
+  @Test
+  def testNoFilterOptions(): Unit = {
+    val offsets = executeAndParse(Array())
+    assertTrue(expectedOffsets() sameElements offsets.filter(r => !isConsumerOffsetTopicPartition(r)))
+    assertEquals(offsetTopicPartitionCount, offsets.count(isConsumerOffsetTopicPartition))
+  }
+
+  @Test
+  def testInternalExcluded(): Unit = {
+    val offsets = executeAndParse(Array("--exclude-internal-topics"))
+    assertTrue(expectedOffsets() sameElements offsets.filter(r => !isConsumerOffsetTopicPartition(r)))
+    assertEquals(0, offsets.count(isConsumerOffsetTopicPartition))
+  }
+
+  @Test
+  def testTopicNameArg(): Unit = {
+    Range(1, topicCount + 1).foreach(i => {
+      val offsets = executeAndParse(Array("--topic", topicName(i)))
+      assertTrue("Offset output did not match for " + topicName(i), expectedOffsetsForTopic(i) sameElements offsets)
+    })
+  }
+
+  @Test
+  def testTopicPatternArg(): Unit = {
+    val offsets = executeAndParse(Array("--topic", "topic.*"))
+    assertTrue(expectedOffsets() sameElements offsets)
+  }
+
+  @Test
+  def testPartitionsArg(): Unit = {
+    val offsets = executeAndParse(Array("--partitions", "0,1"))
+    assertTrue(expectedOffsets().filter(r => r._2 <= 1) sameElements offsets.filter(r => !isConsumerOffsetTopicPartition(r)))
+    assertEquals(2, offsets.count(isConsumerOffsetTopicPartition))
+  }
+
+  @Test
+  def testTopicPatternArgWithPartitionsArg(): Unit = {
+    val offsets = executeAndParse(Array("--topic", "topic.*", "--partitions", "0,1"))
+    assertTrue(expectedOffsets().filter(r => r._2 <= 1) sameElements offsets)
+  }
+
+  @Test
+  def testTopicPartitionsArg(): Unit = {
+    val offsets = executeAndParse(Array("--topic-partitions", "topic1:0,topic2:1,topic(3|4):2,__.*:3"))
+    assertTrue(
+      Array(
+        ("topic1", 0, Some(1)),
+        ("topic2", 1, Some(2)),
+        ("topic3", 2, Some(3)),
+        ("topic4", 2, Some(4))
+      )
+      sameElements offsets.filter(r => !isConsumerOffsetTopicPartition(r))
+    )
+    assertEquals(1, offsets.count(isConsumerOffsetTopicPartition))
+  }
+
+  @Test
+  def testTopicPartitionsArgWithInternalExcluded(): Unit = {
+    val offsets = executeAndParse(Array("--topic-partitions",
+      "topic1:0,topic2:1,topic(3|4):2,__.*:3", "--exclude-internal-topics"))
+    assertTrue(
+      Array(
+        ("topic1", 0, Some(1)),
+        ("topic2", 1, Some(2)),
+        ("topic3", 2, Some(3)),
+        ("topic4", 2, Some(4))
+      )
+        sameElements offsets.filter(r => !isConsumerOffsetTopicPartition(r))
+    )
+    assertEquals(0, offsets.count(isConsumerOffsetTopicPartition))
+  }
+
+  @Test
+  def testTopicPartitionsNotFoundForNonExistentTopic(): Unit = {
+    assertExitCodeIsOne(Array("--topic", "some_nonexistent_topic"))
+  }
+
+  @Test
+  def testTopicPartitionsNotFoundForExcludedInternalTopic(): Unit = {
+    assertExitCodeIsOne(Array("--topic", "some_nonexistent_topic:*"))
+  }
+
+  @Test
+  def testTopicPartitionsNotFoundForNontMatchingTopicPartitionPattern(): Unit = {
+    assertExitCodeIsOne(Array("--topic-partitions", "__consumer_offsets", "--exclude-internal-topics"))
+  }
+
+  private def isConsumerOffsetTopicPartition(record: (String, Int, Option[Long])): Boolean = {
+    record._1 == "__consumer_offsets"
+  }
+
+  private def expectedOffsets(): Array[(String, Int, Option[Long])] = {
+    Range(1, topicCount + 1).flatMap(i => expectedOffsetsForTopic(i)).toArray
+  }
+
+  private def expectedOffsetsForTopic(i: Int): Array[(String, Int, Option[Long])] = {
+    val name = topicName(i)
+    Range(0, i).map(p => (name, p, Some(i.toLong))).toArray
+  }
+
+  private def topicName(i: Int): String = "topic" + i
+
+  private def assertExitCodeIsOne(args: Array[String]): Unit = {
+    var exitStatus: Option[Int] = None
+    Exit.setExitProcedure { (status, _) =>
+      exitStatus = Some(status)
+      throw new RuntimeException
+    }
+
+    try {
+      GetOffsetShell.main(addBootstrapServer(args))
+    } catch {
+      case e: RuntimeException =>
+    } finally {
+      Exit.resetExitProcedure()
+    }
+
+    assertEquals(Some(1), exitStatus)
+  }
+
+  private def executeAndParse(args: Array[String]): Array[(String, Int, Option[Long])] = {
+    val output = executeAndGrabOutput(args)
+    output.split(System.lineSeparator())
+      .map(_.split(":"))
+      .filter(_.length >= 2)
+      .map(line => {
+        val topic = line(0)
+        val partition = line(1).toInt
+        val timestamp = if (line.length == 2 || line(2).isEmpty) None else Some(line(2).toLong)
+        (topic, partition, timestamp)
+      })
+  }
+
+  private def executeAndGrabOutput(args: Array[String]): String = {
+    TestUtils.grabConsoleOutput(GetOffsetShell.main(addBootstrapServer(args)))
+  }
+
+  private def addBootstrapServer(args: Array[String]): Array[String] = {
+    args ++ Array("--bootstrap-server", brokerList)
+  }
+}

From 12901068ff5200c8e8f2d840bbd25373784c6738 Mon Sep 17 00:00:00 2001
From: Daniel Urban <durban@cloudera.com>
Date: Mon, 25 Jan 2021 17:01:28 +0100
Subject: [PATCH 5/7] KAFKA-5235: Addressed review comments

---
 .../scala/kafka/tools/GetOffsetShell.scala    | 146 ++++++-------
 .../tools/GetOffsetShellParsingTest.scala     | 201 ++++++++++++++++++
 .../kafka/tools/GetOffsetShellTest.scala      | 107 ++++++----
 3 files changed, 325 insertions(+), 129 deletions(-)
 create mode 100644 core/src/test/scala/kafka/tools/GetOffsetShellParsingTest.scala

diff --git a/core/src/main/scala/kafka/tools/GetOffsetShell.scala b/core/src/main/scala/kafka/tools/GetOffsetShell.scala
index 54224617fb312..b5b736ce10c89 100644
--- a/core/src/main/scala/kafka/tools/GetOffsetShell.scala
+++ b/core/src/main/scala/kafka/tools/GetOffsetShell.scala
@@ -43,27 +43,25 @@ object GetOffsetShell {
                            .withRequiredArg
                            .describedAs("HOST1:PORT1,...,HOST3:PORT3")
                            .ofType(classOf[String])
-    val topicPartitionOpt = parser.accepts("topic-partitions", "Comma separated list of topic-partition specifications to get the offsets for, with the format of topic:partition. The 'topic' part can be a regex or may be omitted to only specify the partitions, and query all authorized topics." +
+    val topicPartitionsOpt = parser.accepts("topic-partitions", "Comma separated list of topic-partition specifications to get the offsets for, with the format of topic:partition. The 'topic' part can be a regex or may be omitted to only specify the partitions, and query all authorized topics." +
                                             " The 'partition' part 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 of the specified topic.")
                            .withRequiredArg
                            .describedAs("topic:partition,...,topic:partition")
                            .ofType(classOf[String])
-                           .defaultsTo("")
-    val topicOpt = parser.accepts("topic", s"The topic to get the offsets for. It also accepts a regular expression. If not present, all authorized topics are queried. Ignored if $topicPartitionOpt is present.")
+    val topicOpt = parser.accepts("topic", s"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(classOf[String])
-    val partitionOpt = parser.accepts("partitions", s"Comma separated list of partition ids to get the offsets for. If not present, all partitions of the authorized topics are queried. Ignored if $topicPartitionOpt is present.")
+    val partitionsOpt = parser.accepts("partitions", s"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(classOf[String])
-                           .defaultsTo("")
-    val timeOpt = parser.accepts("time", "timestamp of the offsets before that. [Note: No offset is returned, if the timestamp greater than recently commited record timestamp is given.]")
+    val 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(latest)/-2(earliest)")
                            .ofType(classOf[java.lang.Long])
                            .defaultsTo(-1L)
-    val commandConfigOpt = parser.accepts("command-config", s"Consumer config properties file.")
+    val commandConfigOpt = parser.accepts("command-config", s"Property file containing configs to be passed to Consumer Client.")
                            .withRequiredArg
                            .describedAs("config file")
                            .ofType(classOf[String])
@@ -87,9 +85,14 @@ object GetOffsetShell {
     ToolsUtils.validatePortOrDie(parser, brokerList)
     val excludeInternalTopics = options.has(excludeInternalTopicsOpt)
 
+    if (options.has(topicPartitionsOpt) && (options.has(topicOpt) || options.has(partitionsOpt))) {
+      System.err.println(s"--topic-partitions cannot be used with --topic or --partitions")
+      Exit.exit(1)
+    }
+
     val partitionIdsRequested: Set[Int] = {
-      val partitionsString = options.valueOf(partitionOpt)
-      if (partitionsString.isEmpty)
+      val partitionsString = options.valueOf(partitionsOpt)
+      if (partitionsString == null || partitionsString.isEmpty)
         Set.empty
       else
         partitionsString.split(",").map { partitionString =>
@@ -103,8 +106,8 @@ object GetOffsetShell {
     }
     val listOffsetsTimestamp = options.valueOf(timeOpt).longValue
 
-    val topicPartitionFilter = if (options.has(topicPartitionOpt)) {
-      Some(createTopicPartitionFilterWithPatternList(options.valueOf(topicPartitionOpt), excludeInternalTopics))
+    val topicPartitionFilter = if (options.has(topicPartitionsOpt)) {
+      createTopicPartitionFilterWithPatternList(options.valueOf(topicPartitionsOpt), excludeInternalTopics)
     } else {
       createTopicPartitionFilterWithTopicAndPartitionPattern(
         if (options.has(topicOpt)) Some(options.valueOf(topicOpt)) else None,
@@ -128,13 +131,7 @@ object GetOffsetShell {
       Exit.exit(1)
     }
 
-    val topicPartitions = partitionInfos.sortWith((tp1, tp2) => {
-      val topicComp = tp1.topic.compareTo(tp2.topic)
-      if(topicComp == 0)
-        tp1.partition < tp2.partition
-      else
-        topicComp < 0
-    }).flatMap { p =>
+    val topicPartitions = partitionInfos.flatMap { p =>
       if (p.leader == null) {
         System.err.println(s"Error: topic-partition ${p.topic}:${p.partition} does not have a leader. Skip getting offsets")
         None
@@ -162,7 +159,6 @@ object GetOffsetShell {
     }).foreach { case (tp, offset) =>
       println(s"${tp.topic}:${tp.partition}:${Option(offset).getOrElse("")}")
     }
-
   }
 
   /**
@@ -173,26 +169,28 @@ object GetOffsetShell {
    * TopicPattern: REGEX
    * PartitionPattern: NUMBER | NUMBER-(NUMBER)? | -NUMBER
    */
-  private def createTopicPartitionFilterWithPatternList(topicPartitions: String, excludeInternalTopics: Boolean): PartitionInfo => Boolean = {
+  def createTopicPartitionFilterWithPatternList(topicPartitions: String, excludeInternalTopics: Boolean): PartitionInfo => Boolean = {
     val ruleSpecs = topicPartitions.split(",")
-    val rules = ruleSpecs.map(ruleSpec => {
+    val rules = ruleSpecs.map { ruleSpec =>
       val parts = ruleSpec.split(":")
       if (parts.length == 1) {
-        val whitelist = IncludeList(parts(0))
-        tp: PartitionInfo => whitelist.isTopicAllowed(tp.topic, excludeInternalTopics)
+        val includeList = IncludeList(parts(0))
+        tp: PartitionInfo => includeList.isTopicAllowed(tp.topic, excludeInternalTopics)
       } else if (parts.length == 2) {
-        val partitionFilter = createPartitionFilter(parts(1))
-
-        if (parts(0).trim().isEmpty) {
-          tp: PartitionInfo => partitionFilter.apply(tp.partition)
-        } else {
-          val whitelist = IncludeList(parts(0))
-          tp: PartitionInfo => whitelist.isTopicAllowed(tp.topic, excludeInternalTopics) && partitionFilter.apply(tp.partition)
+        try {
+          val partitionFilter = createPartitionFilter(parts(1))
+          val includeList = if (parts(0).trim().isEmpty) IncludeList(".*") else IncludeList(parts(0))
+
+          tp: PartitionInfo => includeList.isTopicAllowed(tp.topic, excludeInternalTopics) && partitionFilter.apply(tp.partition)
+        } catch {
+          case e: IllegalArgumentException =>
+            throw new IllegalArgumentException(s"Invalid rule '$ruleSpec'; Issue: ${e.getMessage}")
         }
+
       } else {
         throw new IllegalArgumentException(s"Invalid topic-partition rule: $ruleSpec")
       }
-    })
+    }
 
     tp => rules.exists(rule => rule.apply(tp))
   }
@@ -202,70 +200,54 @@ object GetOffsetShell {
    * Expected format:
    * PartitionPattern: NUMBER | NUMBER-(NUMBER)? | -NUMBER
    */
-  private def createPartitionFilter(spec: String): Int => Boolean = {
-    if (spec.indexOf('-') != -1) {
-      val rangeParts = spec.split("-", -1)
-      if(rangeParts.length != 2 || rangeParts(0).isEmpty && rangeParts(1).isEmpty) {
-        throw new IllegalArgumentException(s"Invalid range specification: $spec")
-      }
+  def createPartitionFilter(spec: String): Int => Boolean = {
+    try {
+      if (spec.indexOf('-') != -1) {
+        val rangeParts = spec.split("-", -1)
+        if (rangeParts.length != 2 || rangeParts(0).isEmpty && rangeParts(1).isEmpty) {
+          throw new IllegalArgumentException(s"Invalid range specification: $spec")
+        }
 
-      if(rangeParts(0).isEmpty) {
-        val max = rangeParts(1).toInt
-        partition: Int => partition < max
-      } else if(rangeParts(1).isEmpty) {
-        val min = rangeParts(0).toInt
-        partition: Int => partition >= min
-      } else {
-        val min = rangeParts(0).toInt
-        val max = rangeParts(1).toInt
+        if (rangeParts(0).isEmpty) {
+          val max = rangeParts(1).toInt
+          partition: Int => partition < max
+        } else if (rangeParts(1).isEmpty) {
+          val min = rangeParts(0).toInt
+          partition: Int => partition >= min
+        } else {
+          val min = rangeParts(0).toInt
+          val max = rangeParts(1).toInt
 
-        if (min > max) {
-          throw new IllegalArgumentException(s"Range lower bound cannot be greater than upper bound: $spec")
-        }
+          if (min > max) {
+            throw new IllegalArgumentException(s"Range lower bound cannot be greater than upper bound: $spec")
+          }
 
-        partition: Int => partition >= min && partition < max
+          partition: Int => partition >= min && partition < max
+        }
+      } else {
+        val number = spec.toInt
+        partition: Int => partition == number
       }
-    } else {
-      val number = spec.toInt
-      partition: Int => partition == number
+    } catch {
+      case _: NumberFormatException =>
+        throw new IllegalArgumentException(s"Expected a number in partition specification: $spec")
     }
   }
 
   /**
    * Creates a topic-partition filter based on a topic pattern and a set of partition ids.
    */
-  private def createTopicPartitionFilterWithTopicAndPartitionPattern(topicOpt: Option[String], excludeInternalTopics: Boolean, partitionIds: Set[Int]): Option[PartitionInfo => Boolean] = {
-    topicOpt match {
-      case Some(topic) =>
-        val topicsFilter = IncludeList(topic)
-        if(partitionIds.isEmpty)
-          Some(t => topicsFilter.isTopicAllowed(t.topic, excludeInternalTopics))
-        else
-          Some(t => topicsFilter.isTopicAllowed(t.topic, excludeInternalTopics) && partitionIds.contains(t.partition))
-      case None =>
-        if(excludeInternalTopics) {
-          if(partitionIds.isEmpty)
-            Some(t => !Topic.isInternal(t.topic))
-          else
-            Some(t => !Topic.isInternal(t.topic) && partitionIds.contains(t.partition))
-        } else {
-          if(partitionIds.isEmpty)
-            None
-          else
-            Some(t => partitionIds.contains(t.partition))
-        }
-    }
+  def createTopicPartitionFilterWithTopicAndPartitionPattern(topicOpt: Option[String], excludeInternalTopics: Boolean, partitionIds: Set[Int]): PartitionInfo => Boolean = {
+    val topicsFilter = IncludeList(topicOpt.getOrElse(".*"))
+    t => topicsFilter.isTopicAllowed(t.topic, excludeInternalTopics) && (partitionIds.isEmpty || partitionIds.contains(t.partition))
   }
 
   /**
-   * Return the partition infos. Filter them with topicPartitionFilter if specified.
+   * Return the partition infos. Filter them with topicPartitionFilter.
    */
-  private def listPartitionInfos(consumer: KafkaConsumer[_, _], topicPartitionFilter: Option[PartitionInfo => Boolean]): Seq[PartitionInfo] = {
-    val topicListUnfiltered = consumer.listTopics.asScala.values.flatMap { tp => tp.asScala }
-    val topicList = topicPartitionFilter match {
-      case Some(filter) => topicListUnfiltered.filter { tp => filter.apply(tp) }
-      case _ => topicListUnfiltered
-    }
-    topicList.toBuffer
+  private def listPartitionInfos(consumer: KafkaConsumer[_, _], topicPartitionFilter: PartitionInfo => Boolean): Seq[PartitionInfo] = {
+    consumer.listTopics.asScala.values.flatMap { partitions =>
+      partitions.asScala.filter { tp => topicPartitionFilter.apply(tp) }
+    }.toBuffer
   }
 }
diff --git a/core/src/test/scala/kafka/tools/GetOffsetShellParsingTest.scala b/core/src/test/scala/kafka/tools/GetOffsetShellParsingTest.scala
new file mode 100644
index 0000000000000..b49c1a21f8def
--- /dev/null
+++ b/core/src/test/scala/kafka/tools/GetOffsetShellParsingTest.scala
@@ -0,0 +1,201 @@
+/**
+ * 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 kafka.tools
+
+import org.apache.kafka.common.PartitionInfo
+import org.junit.Assert.{assertFalse, assertTrue, assertEquals}
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.junit.runners.Parameterized
+import org.junit.runners.Parameterized.Parameters
+
+import java.util
+
+@RunWith(classOf[Parameterized])
+class GetOffsetShellParsingTest(excludeInternal: Boolean) {
+  @Test
+  def testTopicPartitionFilterForTopicName(): Unit = {
+    val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList("test", excludeInternal)
+    assertTrue(filter.apply(new PartitionInfo("test", 0, null, null, null)))
+    assertTrue(filter.apply(new PartitionInfo("test", 1, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("test1", 0, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("__consumer_offsets", 0, null, null, null)))
+  }
+
+  @Test
+  def testTopicPartitionFilterForInternalTopicName(): Unit = {
+    val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList("__consumer_offsets", excludeInternal)
+    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 0, null, null, null)))
+    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 1, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("test1", 0, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("test2", 0, null, null, null)))
+  }
+
+  @Test
+  def testTopicPartitionFilterForTopicNameList(): Unit = {
+    val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList("test,test1,__consumer_offsets", excludeInternal)
+    assertTrue(filter.apply(new PartitionInfo("test", 0, null, null, null)))
+    assertTrue(filter.apply(new PartitionInfo("test1", 1, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("test2", 0, null, null, null)))
+
+    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 0, null, null, null)))
+  }
+
+  @Test
+  def testTopicPartitionFilterForRegex(): Unit = {
+    val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList("test.*", excludeInternal)
+    assertTrue(filter.apply(new PartitionInfo("test", 0, null, null, null)))
+    assertTrue(filter.apply(new PartitionInfo("test1", 1, null, null, null)))
+    assertTrue(filter.apply(new PartitionInfo("test2", 0, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("__consumer_offsets", 0, null, null, null)))
+  }
+
+  @Test
+  def testTopicPartitionFilterForPartitionIndexSpec(): Unit = {
+    val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList(":0", excludeInternal)
+    assertTrue(filter.apply(new PartitionInfo("test", 0, null, null, null)))
+    assertTrue(filter.apply(new PartitionInfo("test1", 0, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("test2", 1, null, null, null)))
+
+    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 0, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("__consumer_offsets", 1, null, null, null)))
+  }
+
+  @Test
+  def testTopicPartitionFilterForPartitionRangeSpec(): Unit = {
+    val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList(":1-3", excludeInternal)
+    assertTrue(filter.apply(new PartitionInfo("test", 1, null, null, null)))
+    assertTrue(filter.apply(new PartitionInfo("test1", 2, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("test2", 0, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("test2", 3, null, null, null)))
+
+    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 2, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("__consumer_offsets", 3, null, null, null)))
+  }
+
+  @Test
+  def testTopicPartitionFilterForPartitionLowerBoundSpec(): Unit = {
+    val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList(":1-", excludeInternal)
+    assertTrue(filter.apply(new PartitionInfo("test", 1, null, null, null)))
+    assertTrue(filter.apply(new PartitionInfo("test1", 2, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("test2", 0, null, null, null)))
+
+    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 2, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("__consumer_offsets", 0, null, null, null)))
+  }
+
+  @Test
+  def testTopicPartitionFilterForPartitionUpperBoundSpec(): Unit = {
+    val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList(":-3", excludeInternal)
+    assertTrue(filter.apply(new PartitionInfo("test", 0, null, null, null)))
+    assertTrue(filter.apply(new PartitionInfo("test1", 1, null, null, null)))
+    assertTrue(filter.apply(new PartitionInfo("test2", 2, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("test3", 3, null, null, null)))
+
+    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 2, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("__consumer_offsets", 3, null, null, null)))
+  }
+
+  @Test
+  def testTopicPartitionFilterComplex(): Unit = {
+    val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList("test.*:0,__consumer_offsets:1-2,.*:3", excludeInternal)
+    assertTrue(filter.apply(new PartitionInfo("test", 0, null, null, null)))
+    assertTrue(filter.apply(new PartitionInfo("test", 3, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("test", 1, null, null, null)))
+
+    assertTrue(filter.apply(new PartitionInfo("test1", 0, null, null, null)))
+    assertTrue(filter.apply(new PartitionInfo("test1", 3, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("test1", 1, null, null, null)))
+
+    assertTrue(filter.apply(new PartitionInfo("custom", 3, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("custom", 0, null, null, null)))
+
+    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 1, null, null, null)))
+    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 3, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("__consumer_offsets", 0, null, null, null)))
+    assertFalse(filter.apply(new PartitionInfo("__consumer_offsets", 2, null, null, null)))
+  }
+
+  @Test
+  def testPartitionFilterForSingleIndex(): Unit = {
+    val filter = GetOffsetShell.createPartitionFilter("1")
+    assertTrue(filter.apply(1))
+    assertFalse(filter.apply(0))
+    assertFalse(filter.apply(2))
+  }
+
+  @Test
+  def testPartitionFilterForRange(): Unit = {
+    val filter = GetOffsetShell.createPartitionFilter("1-3")
+    assertFalse(filter.apply(0))
+    assertTrue(filter.apply(1))
+    assertTrue(filter.apply(2))
+    assertFalse(filter.apply(3))
+    assertFalse(filter.apply(4))
+    assertFalse(filter.apply(5))
+  }
+
+  @Test
+  def testPartitionFilterForLowerBound(): Unit = {
+    val filter = GetOffsetShell.createPartitionFilter("3-")
+    assertFalse(filter.apply(0))
+    assertFalse(filter.apply(1))
+    assertFalse(filter.apply(2))
+    assertTrue(filter.apply(3))
+    assertTrue(filter.apply(4))
+    assertTrue(filter.apply(5))
+  }
+
+  @Test
+  def testPartitionFilterForUpperBound(): Unit = {
+    val filter = GetOffsetShell.createPartitionFilter("-3")
+    assertTrue(filter.apply(0))
+    assertTrue(filter.apply(1))
+    assertTrue(filter.apply(2))
+    assertFalse(filter.apply(3))
+    assertFalse(filter.apply(4))
+    assertFalse(filter.apply(5))
+  }
+
+  @Test(expected = classOf[IllegalArgumentException])
+  def testPartitionFilterForInvalidSingleIndex(): Unit = {
+    GetOffsetShell.createPartitionFilter("a")
+  }
+
+  @Test(expected = classOf[IllegalArgumentException])
+  def testPartitionFilterForInvalidRange(): Unit = {
+    GetOffsetShell.createPartitionFilter("a-b")
+  }
+
+  @Test(expected = classOf[IllegalArgumentException])
+  def testPartitionFilterForInvalidLowerBound(): Unit = {
+    GetOffsetShell.createPartitionFilter("a-")
+  }
+
+  @Test(expected = classOf[IllegalArgumentException])
+  def testPartitionFilterForInvalidUpperBound(): Unit = {
+    GetOffsetShell.createPartitionFilter("-b")
+  }
+}
+
+object GetOffsetShellParsingTest {
+  @Parameters
+  def parameters: util.Collection[Boolean] = {
+    util.Arrays.asList(true, false)
+  }
+}
diff --git a/core/src/test/scala/kafka/tools/GetOffsetShellTest.scala b/core/src/test/scala/kafka/tools/GetOffsetShellTest.scala
index 7b53733f605ed..0c65b4ef1ae08 100644
--- a/core/src/test/scala/kafka/tools/GetOffsetShellTest.scala
+++ b/core/src/test/scala/kafka/tools/GetOffsetShellTest.scala
@@ -1,41 +1,50 @@
+/**
+ * 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 kafka.tools
 
-import java.time.Duration
 import java.util.Properties
-import java.util.regex.Pattern
-
 import kafka.integration.KafkaServerTestHarness
 import kafka.server.KafkaConfig
 import kafka.utils.{Exit, Logging, TestUtils}
 import org.apache.kafka.clients.CommonClientConfigs
-import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
 import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
-import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}
-import org.junit.Assert.{assertEquals, assertTrue}
+import org.apache.kafka.common.serialization.StringSerializer
+import org.junit.Assert.assertEquals
 import org.junit.{Before, Test}
 
 class GetOffsetShellTest extends KafkaServerTestHarness with Logging {
   private val topicCount = 4
   private val offsetTopicPartitionCount = 4
-  private val topicPattern = Pattern.compile("test.*")
 
   override def generateConfigs: collection.Seq[KafkaConfig] = TestUtils.createBrokerConfigs(1, zkConnect)
-    .map(p => {
-      p.put(KafkaConfig.OffsetsTopicPartitionsProp, offsetTopicPartitionCount)
+    .map { p =>
+      p.put(KafkaConfig.OffsetsTopicPartitionsProp, Int.box(offsetTopicPartitionCount))
       p
-    }).map(KafkaConfig.fromProps)
+    }.map(KafkaConfig.fromProps)
 
   @Before
-  def createTopicAndConsume(): Unit = {
+  def createTestAndInternalTopics(): Unit = {
     Range(1, topicCount + 1).foreach(i => createTopic(topicName(i), i))
 
     val props = new Properties()
     props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerList)
     props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
     props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
-    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer])
-    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer])
-    props.put(ConsumerConfig.GROUP_ID_CONFIG, "GetOffsetShellTest")
 
     // Send X messages to each partition of topicX
     val producer = new KafkaProducer[String, String](props)
@@ -43,84 +52,75 @@ class GetOffsetShellTest extends KafkaServerTestHarness with Logging {
       .foreach(msgCount => producer.send(new ProducerRecord[String, String](topicName(i), msgCount % i, null, "val" + msgCount))))
     producer.close()
 
-    // Consume so consumer offsets topic is created
-    val consumer = new KafkaConsumer[String, String](props)
-    consumer.subscribe(topicPattern)
-    consumer.poll(Duration.ofMillis(1000))
-    consumer.commitSync()
-    consumer.close()
+    TestUtils.createOffsetsTopic(zkClient, servers)
   }
 
   @Test
   def testNoFilterOptions(): Unit = {
     val offsets = executeAndParse(Array())
-    assertTrue(expectedOffsets() sameElements offsets.filter(r => !isConsumerOffsetTopicPartition(r)))
-    assertEquals(offsetTopicPartitionCount, offsets.count(isConsumerOffsetTopicPartition))
+    assertEquals(expectedOffsetsWithInternal(), offsets)
   }
 
   @Test
   def testInternalExcluded(): Unit = {
     val offsets = executeAndParse(Array("--exclude-internal-topics"))
-    assertTrue(expectedOffsets() sameElements offsets.filter(r => !isConsumerOffsetTopicPartition(r)))
-    assertEquals(0, offsets.count(isConsumerOffsetTopicPartition))
+    assertEquals(expectedTestTopicOffsets(), offsets)
   }
 
   @Test
   def testTopicNameArg(): Unit = {
     Range(1, topicCount + 1).foreach(i => {
       val offsets = executeAndParse(Array("--topic", topicName(i)))
-      assertTrue("Offset output did not match for " + topicName(i), expectedOffsetsForTopic(i) sameElements offsets)
+      assertEquals("Offset output did not match for " + topicName(i), expectedOffsetsForTopic(i), offsets)
     })
   }
 
   @Test
   def testTopicPatternArg(): Unit = {
     val offsets = executeAndParse(Array("--topic", "topic.*"))
-    assertTrue(expectedOffsets() sameElements offsets)
+    assertEquals(expectedTestTopicOffsets(), offsets)
   }
 
   @Test
   def testPartitionsArg(): Unit = {
     val offsets = executeAndParse(Array("--partitions", "0,1"))
-    assertTrue(expectedOffsets().filter(r => r._2 <= 1) sameElements offsets.filter(r => !isConsumerOffsetTopicPartition(r)))
-    assertEquals(2, offsets.count(isConsumerOffsetTopicPartition))
+    assertEquals(expectedOffsetsWithInternal().filter { case (_, partition, _) => partition <= 1 }, offsets)
   }
 
   @Test
   def testTopicPatternArgWithPartitionsArg(): Unit = {
     val offsets = executeAndParse(Array("--topic", "topic.*", "--partitions", "0,1"))
-    assertTrue(expectedOffsets().filter(r => r._2 <= 1) sameElements offsets)
+    assertEquals(expectedTestTopicOffsets().filter { case (_, partition, _) => partition <= 1 }, offsets)
   }
 
   @Test
   def testTopicPartitionsArg(): Unit = {
     val offsets = executeAndParse(Array("--topic-partitions", "topic1:0,topic2:1,topic(3|4):2,__.*:3"))
-    assertTrue(
-      Array(
+    assertEquals(
+      List(
+        ("__consumer_offsets", 3, Some(0)),
         ("topic1", 0, Some(1)),
         ("topic2", 1, Some(2)),
         ("topic3", 2, Some(3)),
         ("topic4", 2, Some(4))
-      )
-      sameElements offsets.filter(r => !isConsumerOffsetTopicPartition(r))
+      ),
+      offsets
     )
-    assertEquals(1, offsets.count(isConsumerOffsetTopicPartition))
   }
 
   @Test
   def testTopicPartitionsArgWithInternalExcluded(): Unit = {
     val offsets = executeAndParse(Array("--topic-partitions",
       "topic1:0,topic2:1,topic(3|4):2,__.*:3", "--exclude-internal-topics"))
-    assertTrue(
-      Array(
+    assertEquals(
+      List(
         ("topic1", 0, Some(1)),
         ("topic2", 1, Some(2)),
         ("topic3", 2, Some(3)),
         ("topic4", 2, Some(4))
-      )
-        sameElements offsets.filter(r => !isConsumerOffsetTopicPartition(r))
+      ),
+      offsets
     )
-    assertEquals(0, offsets.count(isConsumerOffsetTopicPartition))
   }
 
   @Test
@@ -134,21 +134,31 @@ class GetOffsetShellTest extends KafkaServerTestHarness with Logging {
   }
 
   @Test
-  def testTopicPartitionsNotFoundForNontMatchingTopicPartitionPattern(): Unit = {
+  def testTopicPartitionsNotFoundForNonMatchingTopicPartitionPattern(): Unit = {
     assertExitCodeIsOne(Array("--topic-partitions", "__consumer_offsets", "--exclude-internal-topics"))
   }
 
-  private def isConsumerOffsetTopicPartition(record: (String, Int, Option[Long])): Boolean = {
-    record._1 == "__consumer_offsets"
+  @Test
+  def testTopicPartitionsFlagWithTopicFlagCauseExit(): Unit = {
+    assertExitCodeIsOne(Array("--topic-partitions", "__consumer_offsets", "--topic", "topic1"))
+  }
+
+  @Test
+  def testTopicPartitionsFlagWithPartitionsFlagCauseExit(): Unit = {
+    assertExitCodeIsOne(Array("--topic-partitions", "__consumer_offsets", "--partitions", "0"))
+  }
+
+  private def expectedOffsetsWithInternal(): List[(String, Int, Option[Long])] = {
+    Range(0, offsetTopicPartitionCount).map(i => ("__consumer_offsets", i, Some(0L))).toList ++ expectedTestTopicOffsets()
   }
 
-  private def expectedOffsets(): Array[(String, Int, Option[Long])] = {
-    Range(1, topicCount + 1).flatMap(i => expectedOffsetsForTopic(i)).toArray
+  private def expectedTestTopicOffsets(): List[(String, Int, Option[Long])] = {
+    Range(1, topicCount + 1).flatMap(i => expectedOffsetsForTopic(i)).toList
   }
 
-  private def expectedOffsetsForTopic(i: Int): Array[(String, Int, Option[Long])] = {
+  private def expectedOffsetsForTopic(i: Int): List[(String, Int, Option[Long])] = {
     val name = topicName(i)
-    Range(0, i).map(p => (name, p, Some(i.toLong))).toArray
+    Range(0, i).map(p => (name, p, Some(i.toLong))).toList
   }
 
   private def topicName(i: Int): String = "topic" + i
@@ -171,7 +181,7 @@ class GetOffsetShellTest extends KafkaServerTestHarness with Logging {
     assertEquals(Some(1), exitStatus)
   }
 
-  private def executeAndParse(args: Array[String]): Array[(String, Int, Option[Long])] = {
+  private def executeAndParse(args: Array[String]): List[(String, Int, Option[Long])] = {
     val output = executeAndGrabOutput(args)
     output.split(System.lineSeparator())
       .map(_.split(":"))
@@ -182,6 +192,7 @@ class GetOffsetShellTest extends KafkaServerTestHarness with Logging {
         val timestamp = if (line.length == 2 || line(2).isEmpty) None else Some(line(2).toLong)
         (topic, partition, timestamp)
       })
+      .toList
   }
 
   private def executeAndGrabOutput(args: Array[String]): String = {
@@ -192,3 +203,5 @@ class GetOffsetShellTest extends KafkaServerTestHarness with Logging {
     args ++ Array("--bootstrap-server", brokerList)
   }
 }
+
+

From 6a9c6648951d9fd5c5bebc4d9c3ca2a0034bb610 Mon Sep 17 00:00:00 2001
From: Daniel Urban <durban@cloudera.com>
Date: Wed, 3 Feb 2021 16:04:44 +0100
Subject: [PATCH 6/7] KAFKA-5235: Rebase, switched pattern parsing to regex,
 removed explicit exits

---
 .../scala/kafka/tools/GetOffsetShell.scala    | 162 ++++++------
 .../tools/GetOffsetShellParsingTest.scala     | 238 +++++++++---------
 .../kafka/tools/GetOffsetShellTest.scala      |  12 +-
 3 files changed, 205 insertions(+), 207 deletions(-)

diff --git a/core/src/main/scala/kafka/tools/GetOffsetShell.scala b/core/src/main/scala/kafka/tools/GetOffsetShell.scala
index b5b736ce10c89..cc6ff44c31db9 100644
--- a/core/src/main/scala/kafka/tools/GetOffsetShell.scala
+++ b/core/src/main/scala/kafka/tools/GetOffsetShell.scala
@@ -20,19 +20,30 @@ package kafka.tools
 
 import java.util.Properties
 import joptsimple._
-import kafka.utils.{CommandLineUtils, Exit, ToolsUtils, IncludeList}
+import kafka.utils.{CommandLineUtils, Exit, IncludeList, ToolsUtils}
 import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
 import org.apache.kafka.common.requests.ListOffsetsRequest
 import org.apache.kafka.common.{PartitionInfo, TopicPartition}
 import org.apache.kafka.common.serialization.ByteArrayDeserializer
 import org.apache.kafka.common.utils.Utils
 
+import java.util.regex.Pattern
 import scala.jdk.CollectionConverters._
 import scala.collection.Seq
+import scala.math.Ordering.Implicits.infixOrderingOps
 
 object GetOffsetShell {
+  private val topicPartitionPattern = Pattern.compile( "([^:,]*)(?::([0-9]*)(?:-([0-9]*))?)?")
 
   def main(args: Array[String]): Unit = {
+    try {
+      fetchOffsets(args)
+    } catch {
+      case e: Exception => Exit.exit(1, Some(e.getMessage))
+    }
+  }
+
+  private def fetchOffsets(args: Array[String]): Unit = {
     val parser = new OptionParser(false)
     val 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
@@ -43,10 +54,11 @@ object GetOffsetShell {
                            .withRequiredArg
                            .describedAs("HOST1:PORT1,...,HOST3:PORT3")
                            .ofType(classOf[String])
-    val topicPartitionsOpt = parser.accepts("topic-partitions", "Comma separated list of topic-partition specifications to get the offsets for, with the format of topic:partition. The 'topic' part can be a regex or may be omitted to only specify the partitions, and query all authorized topics." +
-                                            " The 'partition' part 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 of the specified topic.")
+    val topicPartitionsOpt = parser.accepts("topic-partitions", s"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("topic:partition,...,topic:partition")
+                           .describedAs("topic1:1,topic2:0-3,topic3,topic4:5-,topic5:-3")
                            .ofType(classOf[String])
     val topicOpt = parser.accepts("topic", s"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
@@ -86,29 +98,29 @@ object GetOffsetShell {
     val excludeInternalTopics = options.has(excludeInternalTopicsOpt)
 
     if (options.has(topicPartitionsOpt) && (options.has(topicOpt) || options.has(partitionsOpt))) {
-      System.err.println(s"--topic-partitions cannot be used with --topic or --partitions")
-      Exit.exit(1)
+      throw new IllegalArgumentException("--topic-partitions cannot be used with --topic or --partitions")
     }
 
-    val partitionIdsRequested: Set[Int] = {
-      val partitionsString = options.valueOf(partitionsOpt)
-      if (partitionsString == null || partitionsString.isEmpty)
-        Set.empty
-      else
-        partitionsString.split(",").map { partitionString =>
-          try partitionString.toInt
-          catch {
-            case _: NumberFormatException =>
-              System.err.println(s"--partitions expects a comma separated list of numeric partition ids, but received: $partitionsString")
-              Exit.exit(1)
-          }
-        }.toSet
-    }
     val listOffsetsTimestamp = options.valueOf(timeOpt).longValue
 
     val topicPartitionFilter = if (options.has(topicPartitionsOpt)) {
       createTopicPartitionFilterWithPatternList(options.valueOf(topicPartitionsOpt), excludeInternalTopics)
     } else {
+      val partitionIdsRequested: Set[Int] = {
+        val partitionsString = options.valueOf(partitionsOpt)
+        if (partitionsString == null || partitionsString.isEmpty)
+          Set.empty
+        else
+          partitionsString.split(",").map { partitionString =>
+            try partitionString.toInt
+            catch {
+              case _: NumberFormatException =>
+                throw new IllegalArgumentException(s"--partitions expects a comma separated list of numeric " +
+                  s"partition ids, but received: $partitionsString")
+            }
+          }.toSet
+      }
+
       createTopicPartitionFilterWithTopicAndPartitionPattern(
         if (options.has(topicOpt)) Some(options.valueOf(topicOpt)) else None,
         excludeInternalTopics,
@@ -127,8 +139,7 @@ object GetOffsetShell {
     val partitionInfos = listPartitionInfos(consumer, topicPartitionFilter)
 
     if (partitionInfos.isEmpty) {
-      System.err.println(s"Could not match any topic-partitions with the specified filters")
-      Exit.exit(1)
+      throw new IllegalArgumentException("Could not match any topic-partitions with the specified filters")
     }
 
     val topicPartitions = partitionInfos.flatMap { p =>
@@ -150,17 +161,15 @@ object GetOffsetShell {
         }
     }
 
-    partitionOffsets.toSeq.sortWith((tp1, tp2) => {
-      val topicComp = tp1._1.topic.compareTo(tp2._1.topic)
-      if (topicComp == 0)
-        tp1._1.partition < tp2._1.partition
-      else
-        topicComp < 0
-    }).foreach { case (tp, offset) =>
-      println(s"${tp.topic}:${tp.partition}:${Option(offset).getOrElse("")}")
+    partitionOffsets.toSeq.sortWith((tp1, tp2) => compareTopicPartitions(tp1._1, tp2._1)).foreach {
+      case (tp, offset) => println(s"${tp.topic}:${tp.partition}:${Option(offset).getOrElse("")}")
     }
   }
 
+  def compareTopicPartitions(a: TopicPartition, b: TopicPartition): Boolean = {
+    (a.topic(), a.partition()) < (b.topic(), b.partition())
+  }
+
   /**
    * Creates a topic-partition filter based on a list of patterns.
    * Expected format:
@@ -171,67 +180,50 @@ object GetOffsetShell {
    */
   def createTopicPartitionFilterWithPatternList(topicPartitions: String, excludeInternalTopics: Boolean): PartitionInfo => Boolean = {
     val ruleSpecs = topicPartitions.split(",")
-    val rules = ruleSpecs.map { ruleSpec =>
-      val parts = ruleSpec.split(":")
-      if (parts.length == 1) {
-        val includeList = IncludeList(parts(0))
-        tp: PartitionInfo => includeList.isTopicAllowed(tp.topic, excludeInternalTopics)
-      } else if (parts.length == 2) {
-        try {
-          val partitionFilter = createPartitionFilter(parts(1))
-          val includeList = if (parts(0).trim().isEmpty) IncludeList(".*") else IncludeList(parts(0))
-
-          tp: PartitionInfo => includeList.isTopicAllowed(tp.topic, excludeInternalTopics) && partitionFilter.apply(tp.partition)
-        } catch {
-          case e: IllegalArgumentException =>
-            throw new IllegalArgumentException(s"Invalid rule '$ruleSpec'; Issue: ${e.getMessage}")
-        }
-
-      } else {
-        throw new IllegalArgumentException(s"Invalid topic-partition rule: $ruleSpec")
-      }
-    }
-
+    val rules = ruleSpecs.map { ruleSpec => parseRuleSpec(ruleSpec, excludeInternalTopics) }
     tp => rules.exists(rule => rule.apply(tp))
   }
 
-  /**
-   * Creates a partition filter based on a single id or a range.
-   * Expected format:
-   * PartitionPattern: NUMBER | NUMBER-(NUMBER)? | -NUMBER
-   */
-  def createPartitionFilter(spec: String): Int => Boolean = {
-    try {
-      if (spec.indexOf('-') != -1) {
-        val rangeParts = spec.split("-", -1)
-        if (rangeParts.length != 2 || rangeParts(0).isEmpty && rangeParts(1).isEmpty) {
-          throw new IllegalArgumentException(s"Invalid range specification: $spec")
-        }
+  def parseRuleSpec(ruleSpec: String, excludeInternalTopics: Boolean): PartitionInfo => Boolean = {
+    def wrapNullOrEmpty(s: String): Option[String] = {
+      if (s == null || s.isEmpty)
+        None
+      else
+        Some(s)
+    }
 
-        if (rangeParts(0).isEmpty) {
-          val max = rangeParts(1).toInt
-          partition: Int => partition < max
-        } else if (rangeParts(1).isEmpty) {
-          val min = rangeParts(0).toInt
-          partition: Int => partition >= min
-        } else {
-          val min = rangeParts(0).toInt
-          val max = rangeParts(1).toInt
-
-          if (min > max) {
-            throw new IllegalArgumentException(s"Range lower bound cannot be greater than upper bound: $spec")
-          }
-
-          partition: Int => partition >= min && partition < max
-        }
-      } else {
-        val number = spec.toInt
-        partition: Int => partition == number
+    val matcher = topicPartitionPattern.matcher(ruleSpec)
+    if (!matcher.matches() || matcher.groupCount() == 0)
+      throw new IllegalArgumentException(s"Invalid rule specification: $ruleSpec")
+
+    val topicPattern = wrapNullOrEmpty(matcher.group(1))
+    val lowerRange = if (matcher.groupCount() >= 2) wrapNullOrEmpty(matcher.group(2)) else None
+    val upperRangeSection = if (matcher.groupCount() >= 3) matcher.group(3) else null
+    val upperRange = wrapNullOrEmpty(upperRangeSection)
+    val isRange = upperRangeSection != null
+
+    val includeList = IncludeList(topicPattern.getOrElse(".*"))
+
+    val partitionFilter: Int => Boolean = if (lowerRange.isEmpty && upperRange.isEmpty) {
+      _ => true
+    } else if (lowerRange.isEmpty) {
+      val upperBound = upperRange.get.toInt
+      p => p < upperBound
+    } else if (upperRange.isEmpty) {
+      val lowerBound = lowerRange.get.toInt
+      if (isRange) {
+        p => p >= lowerBound
       }
-    } catch {
-      case _: NumberFormatException =>
-        throw new IllegalArgumentException(s"Expected a number in partition specification: $spec")
+      else {
+        p => p == lowerBound
+      }
+    } else {
+      val upperBound = upperRange.get.toInt
+      val lowerBound = lowerRange.get.toInt
+      p => p >= lowerBound && p < upperBound
     }
+
+    tp => includeList.isTopicAllowed(tp.topic(), excludeInternalTopics) && partitionFilter.apply(tp.partition())
   }
 
   /**
diff --git a/core/src/test/scala/kafka/tools/GetOffsetShellParsingTest.scala b/core/src/test/scala/kafka/tools/GetOffsetShellParsingTest.scala
index b49c1a21f8def..edfadea401eca 100644
--- a/core/src/test/scala/kafka/tools/GetOffsetShellParsingTest.scala
+++ b/core/src/test/scala/kafka/tools/GetOffsetShellParsingTest.scala
@@ -18,184 +18,190 @@
 package kafka.tools
 
 import org.apache.kafka.common.PartitionInfo
-import org.junit.Assert.{assertFalse, assertTrue, assertEquals}
-import org.junit.Test
-import org.junit.runner.RunWith
-import org.junit.runners.Parameterized
-import org.junit.runners.Parameterized.Parameters
-
-import java.util
-
-@RunWith(classOf[Parameterized])
-class GetOffsetShellParsingTest(excludeInternal: Boolean) {
-  @Test
-  def testTopicPartitionFilterForTopicName(): Unit = {
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertThrows, assertTrue}
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.ValueSource
+
+class GetOffsetShellParsingTest {
+  @ParameterizedTest
+  @ValueSource(booleans = Array(true, false))
+  def testTopicPartitionFilterForTopicName(excludeInternal: Boolean): Unit = {
     val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList("test", excludeInternal)
-    assertTrue(filter.apply(new PartitionInfo("test", 0, null, null, null)))
-    assertTrue(filter.apply(new PartitionInfo("test", 1, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("test1", 0, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("__consumer_offsets", 0, null, null, null)))
+    assertTrue(filter.apply(partitionInfo("test", 0)))
+    assertTrue(filter.apply(partitionInfo("test", 1)))
+    assertFalse(filter.apply(partitionInfo("test1", 0)))
+    assertFalse(filter.apply(partitionInfo("__consumer_offsets", 0)))
   }
 
-  @Test
-  def testTopicPartitionFilterForInternalTopicName(): Unit = {
+  @ParameterizedTest
+  @ValueSource(booleans = Array(true, false))
+  def testTopicPartitionFilterForInternalTopicName(excludeInternal: Boolean): Unit = {
     val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList("__consumer_offsets", excludeInternal)
-    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 0, null, null, null)))
-    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 1, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("test1", 0, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("test2", 0, null, null, null)))
+    assertEquals(!excludeInternal, filter.apply(partitionInfo("__consumer_offsets", 0)))
+    assertEquals(!excludeInternal, filter.apply(partitionInfo("__consumer_offsets", 1)))
+    assertFalse(filter.apply(partitionInfo("test1", 0)))
+    assertFalse(filter.apply(partitionInfo("test2", 0)))
   }
 
-  @Test
-  def testTopicPartitionFilterForTopicNameList(): Unit = {
+  @ParameterizedTest
+  @ValueSource(booleans = Array(true, false))
+  def testTopicPartitionFilterForTopicNameList(excludeInternal: Boolean): Unit = {
     val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList("test,test1,__consumer_offsets", excludeInternal)
-    assertTrue(filter.apply(new PartitionInfo("test", 0, null, null, null)))
-    assertTrue(filter.apply(new PartitionInfo("test1", 1, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("test2", 0, null, null, null)))
+    assertTrue(filter.apply(partitionInfo("test", 0)))
+    assertTrue(filter.apply(partitionInfo("test1", 1)))
+    assertFalse(filter.apply(partitionInfo("test2", 0)))
 
-    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 0, null, null, null)))
+    assertEquals(!excludeInternal, filter.apply(partitionInfo("__consumer_offsets", 0)))
   }
 
-  @Test
-  def testTopicPartitionFilterForRegex(): Unit = {
+  @ParameterizedTest
+  @ValueSource(booleans = Array(true, false))
+  def testTopicPartitionFilterForRegex(excludeInternal: Boolean): Unit = {
     val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList("test.*", excludeInternal)
-    assertTrue(filter.apply(new PartitionInfo("test", 0, null, null, null)))
-    assertTrue(filter.apply(new PartitionInfo("test1", 1, null, null, null)))
-    assertTrue(filter.apply(new PartitionInfo("test2", 0, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("__consumer_offsets", 0, null, null, null)))
+    assertTrue(filter.apply(partitionInfo("test", 0)))
+    assertTrue(filter.apply(partitionInfo("test1", 1)))
+    assertTrue(filter.apply(partitionInfo("test2", 0)))
+    assertFalse(filter.apply(partitionInfo("__consumer_offsets", 0)))
   }
 
-  @Test
-  def testTopicPartitionFilterForPartitionIndexSpec(): Unit = {
+  @ParameterizedTest
+  @ValueSource(booleans = Array(true, false))
+  def testTopicPartitionFilterForPartitionIndexSpec(excludeInternal: Boolean): Unit = {
     val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList(":0", excludeInternal)
-    assertTrue(filter.apply(new PartitionInfo("test", 0, null, null, null)))
-    assertTrue(filter.apply(new PartitionInfo("test1", 0, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("test2", 1, null, null, null)))
+    assertTrue(filter.apply(partitionInfo("test", 0)))
+    assertTrue(filter.apply(partitionInfo("test1", 0)))
+    assertFalse(filter.apply(partitionInfo("test2", 1)))
 
-    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 0, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("__consumer_offsets", 1, null, null, null)))
+    assertEquals(!excludeInternal, filter.apply(partitionInfo("__consumer_offsets", 0)))
+    assertFalse(filter.apply(partitionInfo("__consumer_offsets", 1)))
   }
 
-  @Test
-  def testTopicPartitionFilterForPartitionRangeSpec(): Unit = {
+  @ParameterizedTest
+  @ValueSource(booleans = Array(true, false))
+  def testTopicPartitionFilterForPartitionRangeSpec(excludeInternal: Boolean): Unit = {
     val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList(":1-3", excludeInternal)
-    assertTrue(filter.apply(new PartitionInfo("test", 1, null, null, null)))
-    assertTrue(filter.apply(new PartitionInfo("test1", 2, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("test2", 0, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("test2", 3, null, null, null)))
+    assertTrue(filter.apply(partitionInfo("test", 1)))
+    assertTrue(filter.apply(partitionInfo("test1", 2)))
+    assertFalse(filter.apply(partitionInfo("test2", 0)))
+    assertFalse(filter.apply(partitionInfo("test2", 3)))
 
-    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 2, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("__consumer_offsets", 3, null, null, null)))
+    assertEquals(!excludeInternal, filter.apply(partitionInfo("__consumer_offsets", 2)))
+    assertFalse(filter.apply(partitionInfo("__consumer_offsets", 3)))
   }
 
-  @Test
-  def testTopicPartitionFilterForPartitionLowerBoundSpec(): Unit = {
+  @ParameterizedTest
+  @ValueSource(booleans = Array(true, false))
+  def testTopicPartitionFilterForPartitionLowerBoundSpec(excludeInternal: Boolean): Unit = {
     val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList(":1-", excludeInternal)
-    assertTrue(filter.apply(new PartitionInfo("test", 1, null, null, null)))
-    assertTrue(filter.apply(new PartitionInfo("test1", 2, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("test2", 0, null, null, null)))
+    assertTrue(filter.apply(partitionInfo("test", 1)))
+    assertTrue(filter.apply(partitionInfo("test1", 2)))
+    assertFalse(filter.apply(partitionInfo("test2", 0)))
 
-    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 2, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("__consumer_offsets", 0, null, null, null)))
+    assertEquals(!excludeInternal, filter.apply(partitionInfo("__consumer_offsets", 2)))
+    assertFalse(filter.apply(partitionInfo("__consumer_offsets", 0)))
   }
 
-  @Test
-  def testTopicPartitionFilterForPartitionUpperBoundSpec(): Unit = {
+  @ParameterizedTest
+  @ValueSource(booleans = Array(true, false))
+  def testTopicPartitionFilterForPartitionUpperBoundSpec(excludeInternal: Boolean): Unit = {
     val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList(":-3", excludeInternal)
-    assertTrue(filter.apply(new PartitionInfo("test", 0, null, null, null)))
-    assertTrue(filter.apply(new PartitionInfo("test1", 1, null, null, null)))
-    assertTrue(filter.apply(new PartitionInfo("test2", 2, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("test3", 3, null, null, null)))
+    assertTrue(filter.apply(partitionInfo("test", 0)))
+    assertTrue(filter.apply(partitionInfo("test1", 1)))
+    assertTrue(filter.apply(partitionInfo("test2", 2)))
+    assertFalse(filter.apply(partitionInfo("test3", 3)))
 
-    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 2, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("__consumer_offsets", 3, null, null, null)))
+    assertEquals(!excludeInternal, filter.apply(partitionInfo("__consumer_offsets", 2)))
+    assertFalse(filter.apply(partitionInfo("__consumer_offsets", 3)))
   }
 
-  @Test
-  def testTopicPartitionFilterComplex(): Unit = {
+  @ParameterizedTest
+  @ValueSource(booleans = Array(true, false))
+  def testTopicPartitionFilterComplex(excludeInternal: Boolean): Unit = {
     val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList("test.*:0,__consumer_offsets:1-2,.*:3", excludeInternal)
-    assertTrue(filter.apply(new PartitionInfo("test", 0, null, null, null)))
-    assertTrue(filter.apply(new PartitionInfo("test", 3, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("test", 1, null, null, null)))
+    assertTrue(filter.apply(partitionInfo("test", 0)))
+    assertTrue(filter.apply(partitionInfo("test", 3)))
+    assertFalse(filter.apply(partitionInfo("test", 1)))
 
-    assertTrue(filter.apply(new PartitionInfo("test1", 0, null, null, null)))
-    assertTrue(filter.apply(new PartitionInfo("test1", 3, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("test1", 1, null, null, null)))
+    assertTrue(filter.apply(partitionInfo("test1", 0)))
+    assertTrue(filter.apply(partitionInfo("test1", 3)))
+    assertFalse(filter.apply(partitionInfo("test1", 1)))
 
-    assertTrue(filter.apply(new PartitionInfo("custom", 3, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("custom", 0, null, null, null)))
+    assertTrue(filter.apply(partitionInfo("custom", 3)))
+    assertFalse(filter.apply(partitionInfo("custom", 0)))
 
-    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 1, null, null, null)))
-    assertEquals(!excludeInternal, filter.apply(new PartitionInfo("__consumer_offsets", 3, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("__consumer_offsets", 0, null, null, null)))
-    assertFalse(filter.apply(new PartitionInfo("__consumer_offsets", 2, null, null, null)))
+    assertEquals(!excludeInternal, filter.apply(partitionInfo("__consumer_offsets", 1)))
+    assertEquals(!excludeInternal, filter.apply(partitionInfo("__consumer_offsets", 3)))
+    assertFalse(filter.apply(partitionInfo("__consumer_offsets", 0)))
+    assertFalse(filter.apply(partitionInfo("__consumer_offsets", 2)))
   }
 
   @Test
   def testPartitionFilterForSingleIndex(): Unit = {
-    val filter = GetOffsetShell.createPartitionFilter("1")
-    assertTrue(filter.apply(1))
-    assertFalse(filter.apply(0))
-    assertFalse(filter.apply(2))
+    val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList(":1", excludeInternalTopics = false)
+    assertTrue(filter.apply(partitionInfo("test", 1)))
+    assertFalse(filter.apply(partitionInfo("test", 0)))
+    assertFalse(filter.apply(partitionInfo("test", 2)))
   }
 
   @Test
   def testPartitionFilterForRange(): Unit = {
-    val filter = GetOffsetShell.createPartitionFilter("1-3")
-    assertFalse(filter.apply(0))
-    assertTrue(filter.apply(1))
-    assertTrue(filter.apply(2))
-    assertFalse(filter.apply(3))
-    assertFalse(filter.apply(4))
-    assertFalse(filter.apply(5))
+    val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList(":1-3", excludeInternalTopics = false)
+    assertFalse(filter.apply(partitionInfo("test", 0)))
+    assertTrue(filter.apply(partitionInfo("test", 1)))
+    assertTrue(filter.apply(partitionInfo("test", 2)))
+    assertFalse(filter.apply(partitionInfo("test", 3)))
+    assertFalse(filter.apply(partitionInfo("test", 4)))
+    assertFalse(filter.apply(partitionInfo("test", 5)))
   }
 
   @Test
   def testPartitionFilterForLowerBound(): Unit = {
-    val filter = GetOffsetShell.createPartitionFilter("3-")
-    assertFalse(filter.apply(0))
-    assertFalse(filter.apply(1))
-    assertFalse(filter.apply(2))
-    assertTrue(filter.apply(3))
-    assertTrue(filter.apply(4))
-    assertTrue(filter.apply(5))
+    val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList(":3-", excludeInternalTopics = false)
+    assertFalse(filter.apply(partitionInfo("test", 0)))
+    assertFalse(filter.apply(partitionInfo("test", 1)))
+    assertFalse(filter.apply(partitionInfo("test", 2)))
+    assertTrue(filter.apply(partitionInfo("test", 3)))
+    assertTrue(filter.apply(partitionInfo("test", 4)))
+    assertTrue(filter.apply(partitionInfo("test", 5)))
   }
 
   @Test
   def testPartitionFilterForUpperBound(): Unit = {
-    val filter = GetOffsetShell.createPartitionFilter("-3")
-    assertTrue(filter.apply(0))
-    assertTrue(filter.apply(1))
-    assertTrue(filter.apply(2))
-    assertFalse(filter.apply(3))
-    assertFalse(filter.apply(4))
-    assertFalse(filter.apply(5))
+    val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList(":-3", excludeInternalTopics = false)
+    assertTrue(filter.apply(partitionInfo("test", 0)))
+    assertTrue(filter.apply(partitionInfo("test", 1)))
+    assertTrue(filter.apply(partitionInfo("test", 2)))
+    assertFalse(filter.apply(partitionInfo("test", 3)))
+    assertFalse(filter.apply(partitionInfo("test", 4)))
+    assertFalse(filter.apply(partitionInfo("test", 5)))
   }
 
-  @Test(expected = classOf[IllegalArgumentException])
+  @Test
   def testPartitionFilterForInvalidSingleIndex(): Unit = {
-    GetOffsetShell.createPartitionFilter("a")
+    assertThrows(classOf[IllegalArgumentException],
+      () => GetOffsetShell.createTopicPartitionFilterWithPatternList(":a", excludeInternalTopics = false))
   }
 
-  @Test(expected = classOf[IllegalArgumentException])
+  @Test
   def testPartitionFilterForInvalidRange(): Unit = {
-    GetOffsetShell.createPartitionFilter("a-b")
+    assertThrows(classOf[IllegalArgumentException],
+      () => GetOffsetShell.createTopicPartitionFilterWithPatternList(":a-b", excludeInternalTopics = false))
   }
 
-  @Test(expected = classOf[IllegalArgumentException])
+  @Test
   def testPartitionFilterForInvalidLowerBound(): Unit = {
-    GetOffsetShell.createPartitionFilter("a-")
+    assertThrows(classOf[IllegalArgumentException],
+      () => GetOffsetShell.createTopicPartitionFilterWithPatternList(":a-", excludeInternalTopics = false))
   }
 
-  @Test(expected = classOf[IllegalArgumentException])
+  @Test
   def testPartitionFilterForInvalidUpperBound(): Unit = {
-    GetOffsetShell.createPartitionFilter("-b")
+    assertThrows(classOf[IllegalArgumentException],
+      () => GetOffsetShell.createTopicPartitionFilterWithPatternList(":-b", excludeInternalTopics = false))
   }
-}
 
-object GetOffsetShellParsingTest {
-  @Parameters
-  def parameters: util.Collection[Boolean] = {
-    util.Arrays.asList(true, false)
+  private def partitionInfo(topic: String, partition: Int): PartitionInfo = {
+    new PartitionInfo(topic, partition, null, null, null)
   }
 }
diff --git a/core/src/test/scala/kafka/tools/GetOffsetShellTest.scala b/core/src/test/scala/kafka/tools/GetOffsetShellTest.scala
index 0c65b4ef1ae08..cd611de177305 100644
--- a/core/src/test/scala/kafka/tools/GetOffsetShellTest.scala
+++ b/core/src/test/scala/kafka/tools/GetOffsetShellTest.scala
@@ -24,8 +24,8 @@ import kafka.utils.{Exit, Logging, TestUtils}
 import org.apache.kafka.clients.CommonClientConfigs
 import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
 import org.apache.kafka.common.serialization.StringSerializer
-import org.junit.Assert.assertEquals
-import org.junit.{Before, Test}
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.{BeforeEach, Test}
 
 class GetOffsetShellTest extends KafkaServerTestHarness with Logging {
   private val topicCount = 4
@@ -37,7 +37,7 @@ class GetOffsetShellTest extends KafkaServerTestHarness with Logging {
       p
     }.map(KafkaConfig.fromProps)
 
-  @Before
+  @BeforeEach
   def createTestAndInternalTopics(): Unit = {
     Range(1, topicCount + 1).foreach(i => createTopic(topicName(i), i))
 
@@ -71,7 +71,7 @@ class GetOffsetShellTest extends KafkaServerTestHarness with Logging {
   def testTopicNameArg(): Unit = {
     Range(1, topicCount + 1).foreach(i => {
       val offsets = executeAndParse(Array("--topic", topicName(i)))
-      assertEquals("Offset output did not match for " + topicName(i), expectedOffsetsForTopic(i), offsets)
+      assertEquals(expectedOffsetsForTopic(i), offsets, () => "Offset output did not match for " + topicName(i))
     })
   }
 
@@ -186,12 +186,12 @@ class GetOffsetShellTest extends KafkaServerTestHarness with Logging {
     output.split(System.lineSeparator())
       .map(_.split(":"))
       .filter(_.length >= 2)
-      .map(line => {
+      .map { line =>
         val topic = line(0)
         val partition = line(1).toInt
         val timestamp = if (line.length == 2 || line(2).isEmpty) None else Some(line(2).toLong)
         (topic, partition, timestamp)
-      })
+      }
       .toList
   }
 

From fabd539f30737ba7c6d163a7a58519e4d94d732a Mon Sep 17 00:00:00 2001
From: Daniel Urban <durban@cloudera.com>
Date: Tue, 9 Feb 2021 09:20:30 +0100
Subject: [PATCH 7/7] KAFKA-5235: Cleanup based on review

---
 .../scala/kafka/tools/GetOffsetShell.scala    | 143 ++++++++----------
 tests/kafkatest/services/kafka/kafka.py       |   2 +-
 2 files changed, 66 insertions(+), 79 deletions(-)

diff --git a/core/src/main/scala/kafka/tools/GetOffsetShell.scala b/core/src/main/scala/kafka/tools/GetOffsetShell.scala
index cc6ff44c31db9..dfd5a227689ec 100644
--- a/core/src/main/scala/kafka/tools/GetOffsetShell.scala
+++ b/core/src/main/scala/kafka/tools/GetOffsetShell.scala
@@ -33,13 +33,15 @@ import scala.collection.Seq
 import scala.math.Ordering.Implicits.infixOrderingOps
 
 object GetOffsetShell {
-  private val topicPartitionPattern = Pattern.compile( "([^:,]*)(?::([0-9]*)(?:-([0-9]*))?)?")
+  private val TopicPartitionPattern = Pattern.compile("([^:,]*)(?::(?:([0-9]*)|(?:([0-9]*)-([0-9]*))))?")
 
   def main(args: Array[String]): Unit = {
     try {
       fetchOffsets(args)
     } catch {
-      case e: Exception => Exit.exit(1, Some(e.getMessage))
+      case e: Exception =>
+        println(s"Error occurred: ${e.getMessage}")
+        Exit.exit(1, Some(e.getMessage))
     }
   }
 
@@ -54,7 +56,7 @@ object GetOffsetShell {
                            .withRequiredArg
                            .describedAs("HOST1:PORT1,...,HOST3:PORT3")
                            .ofType(classOf[String])
-    val topicPartitionsOpt = parser.accepts("topic-partitions", s"Comma separated list of topic-partition patterns to get the offsets for, with the format of '$topicPartitionPattern'." +
+    val topicPartitionsOpt = parser.accepts("topic-partitions", s"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
@@ -106,20 +108,7 @@ object GetOffsetShell {
     val topicPartitionFilter = if (options.has(topicPartitionsOpt)) {
       createTopicPartitionFilterWithPatternList(options.valueOf(topicPartitionsOpt), excludeInternalTopics)
     } else {
-      val partitionIdsRequested: Set[Int] = {
-        val partitionsString = options.valueOf(partitionsOpt)
-        if (partitionsString == null || partitionsString.isEmpty)
-          Set.empty
-        else
-          partitionsString.split(",").map { partitionString =>
-            try partitionString.toInt
-            catch {
-              case _: NumberFormatException =>
-                throw new IllegalArgumentException(s"--partitions expects a comma separated list of numeric " +
-                  s"partition ids, but received: $partitionsString")
-            }
-          }.toSet
-      }
+      val partitionIdsRequested = createPartitionSet(options.valueOf(partitionsOpt))
 
       createTopicPartitionFilterWithTopicAndPartitionPattern(
         if (options.has(topicOpt)) Some(options.valueOf(topicOpt)) else None,
@@ -136,33 +125,37 @@ object GetOffsetShell {
     config.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, clientId)
     val consumer = new KafkaConsumer(config, new ByteArrayDeserializer, new ByteArrayDeserializer)
 
-    val partitionInfos = listPartitionInfos(consumer, topicPartitionFilter)
+    try {
+      val partitionInfos = listPartitionInfos(consumer, topicPartitionFilter)
 
-    if (partitionInfos.isEmpty) {
-      throw new IllegalArgumentException("Could not match any topic-partitions with the specified filters")
-    }
+      if (partitionInfos.isEmpty) {
+        throw new IllegalArgumentException("Could not match any topic-partitions with the specified filters")
+      }
 
-    val topicPartitions = partitionInfos.flatMap { p =>
-      if (p.leader == null) {
-        System.err.println(s"Error: topic-partition ${p.topic}:${p.partition} does not have a leader. Skip getting offsets")
-        None
-      } else
-        Some(new TopicPartition(p.topic, p.partition))
-    }
+      val topicPartitions = partitionInfos.flatMap { p =>
+        if (p.leader == null) {
+          System.err.println(s"Error: topic-partition ${p.topic}:${p.partition} does not have a leader. Skip getting offsets")
+          None
+        } else
+          Some(new TopicPartition(p.topic, p.partition))
+      }
 
-    /* Note that the value of the map can be null */
-    val partitionOffsets: collection.Map[TopicPartition, java.lang.Long] = listOffsetsTimestamp match {
-      case ListOffsetsRequest.EARLIEST_TIMESTAMP => consumer.beginningOffsets(topicPartitions.asJava).asScala
-      case ListOffsetsRequest.LATEST_TIMESTAMP => consumer.endOffsets(topicPartitions.asJava).asScala
-      case _ =>
-        val timestampsToSearch = topicPartitions.map(tp => tp -> (listOffsetsTimestamp: java.lang.Long)).toMap.asJava
-        consumer.offsetsForTimes(timestampsToSearch).asScala.map { case (k, x) =>
-          if (x == null) (k, null) else (k, x.offset: java.lang.Long)
-        }
-    }
+      /* Note that the value of the map can be null */
+      val partitionOffsets: collection.Map[TopicPartition, java.lang.Long] = listOffsetsTimestamp match {
+        case ListOffsetsRequest.EARLIEST_TIMESTAMP => consumer.beginningOffsets(topicPartitions.asJava).asScala
+        case ListOffsetsRequest.LATEST_TIMESTAMP => consumer.endOffsets(topicPartitions.asJava).asScala
+        case _ =>
+          val timestampsToSearch = topicPartitions.map(tp => tp -> (listOffsetsTimestamp: java.lang.Long)).toMap.asJava
+          consumer.offsetsForTimes(timestampsToSearch).asScala.map { case (k, x) =>
+            if (x == null) (k, null) else (k, x.offset: java.lang.Long)
+          }
+      }
 
-    partitionOffsets.toSeq.sortWith((tp1, tp2) => compareTopicPartitions(tp1._1, tp2._1)).foreach {
-      case (tp, offset) => println(s"${tp.topic}:${tp.partition}:${Option(offset).getOrElse("")}")
+      partitionOffsets.toSeq.sortWith((tp1, tp2) => compareTopicPartitions(tp1._1, tp2._1)).foreach {
+        case (tp, offset) => println(s"${tp.topic}:${tp.partition}:${Option(offset).getOrElse("")}")
+      }
+    } finally {
+      consumer.close()
     }
   }
 
@@ -180,50 +173,30 @@ object GetOffsetShell {
    */
   def createTopicPartitionFilterWithPatternList(topicPartitions: String, excludeInternalTopics: Boolean): PartitionInfo => Boolean = {
     val ruleSpecs = topicPartitions.split(",")
-    val rules = ruleSpecs.map { ruleSpec => parseRuleSpec(ruleSpec, excludeInternalTopics) }
-    tp => rules.exists(rule => rule.apply(tp))
+    val rules = ruleSpecs.map(ruleSpec => parseRuleSpec(ruleSpec, excludeInternalTopics))
+    tp => rules.exists { rule => rule.apply(tp) }
   }
 
   def parseRuleSpec(ruleSpec: String, excludeInternalTopics: Boolean): PartitionInfo => Boolean = {
-    def wrapNullOrEmpty(s: String): Option[String] = {
-      if (s == null || s.isEmpty)
-        None
-      else
-        Some(s)
-    }
-
-    val matcher = topicPartitionPattern.matcher(ruleSpec)
-    if (!matcher.matches() || matcher.groupCount() == 0)
+    val matcher = TopicPartitionPattern.matcher(ruleSpec)
+    if (!matcher.matches())
       throw new IllegalArgumentException(s"Invalid rule specification: $ruleSpec")
 
-    val topicPattern = wrapNullOrEmpty(matcher.group(1))
-    val lowerRange = if (matcher.groupCount() >= 2) wrapNullOrEmpty(matcher.group(2)) else None
-    val upperRangeSection = if (matcher.groupCount() >= 3) matcher.group(3) else null
-    val upperRange = wrapNullOrEmpty(upperRangeSection)
-    val isRange = upperRangeSection != null
-
-    val includeList = IncludeList(topicPattern.getOrElse(".*"))
-
-    val partitionFilter: Int => Boolean = if (lowerRange.isEmpty && upperRange.isEmpty) {
-      _ => true
-    } else if (lowerRange.isEmpty) {
-      val upperBound = upperRange.get.toInt
-      p => p < upperBound
-    } else if (upperRange.isEmpty) {
-      val lowerBound = lowerRange.get.toInt
-      if (isRange) {
-        p => p >= lowerBound
-      }
-      else {
-        p => p == lowerBound
-      }
-    } else {
-      val upperBound = upperRange.get.toInt
-      val lowerBound = lowerRange.get.toInt
-      p => p >= lowerBound && p < upperBound
+    def group(group: Int): Option[String] = {
+      Option(matcher.group(group)).filter(s => s != null && s.nonEmpty)
+    }
+
+    val topicFilter = IncludeList(group(1).getOrElse(".*"))
+    val partitionFilter = group(2).map(_.toInt) match {
+      case Some(partition) =>
+        (p: Int) => p == partition
+      case None =>
+        val lowerRange = group(3).map(_.toInt).getOrElse(0)
+        val upperRange = group(4).map(_.toInt).getOrElse(Int.MaxValue)
+        (p: Int) => p >= lowerRange && p < upperRange
     }
 
-    tp => includeList.isTopicAllowed(tp.topic(), excludeInternalTopics) && partitionFilter.apply(tp.partition())
+    tp => topicFilter.isTopicAllowed(tp.topic, excludeInternalTopics) && partitionFilter(tp.partition)
   }
 
   /**
@@ -234,12 +207,26 @@ object GetOffsetShell {
     t => topicsFilter.isTopicAllowed(t.topic, excludeInternalTopics) && (partitionIds.isEmpty || partitionIds.contains(t.partition))
   }
 
+  def createPartitionSet(partitionsString: String): Set[Int] = {
+    if (partitionsString == null || partitionsString.isEmpty)
+      Set.empty
+    else
+      partitionsString.split(",").map { partitionString =>
+        try partitionString.toInt
+        catch {
+          case _: NumberFormatException =>
+            throw new IllegalArgumentException(s"--partitions expects a comma separated list of numeric " +
+              s"partition ids, but received: $partitionsString")
+        }
+      }.toSet
+  }
+
   /**
    * Return the partition infos. Filter them with topicPartitionFilter.
    */
   private def listPartitionInfos(consumer: KafkaConsumer[_, _], topicPartitionFilter: PartitionInfo => Boolean): Seq[PartitionInfo] = {
     consumer.listTopics.asScala.values.flatMap { partitions =>
-      partitions.asScala.filter { tp => topicPartitionFilter.apply(tp) }
+      partitions.asScala.filter(topicPartitionFilter)
     }.toBuffer
   }
 }
diff --git a/tests/kafkatest/services/kafka/kafka.py b/tests/kafkatest/services/kafka/kafka.py
index 40592f67be79d..4da2ab2e183d0 100644
--- a/tests/kafkatest/services/kafka/kafka.py
+++ b/tests/kafkatest/services/kafka/kafka.py
@@ -1158,7 +1158,7 @@ def get_offset_shell(self, time=None, topic=None, partitions=None, topic_partiti
         cmd = fix_opts_for_new_jvm(node)
         cmd += self.path.script("kafka-run-class.sh", node)
         cmd += " kafka.tools.GetOffsetShell"
-        cmd += " --boostrap-server %s" % self.bootstrap_servers(self.security_protocol)
+        cmd += " --bootstrap-server %s" % self.bootstrap_servers(self.security_protocol)
 
         if time:
             cmd += ' --time %s' % time