You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/02/03 15:15:03 UTC
[GitHub] [kafka] urbandan commented on a change in pull request #9430: KAFKA-5235: GetOffsetShell: support for multiple topics and consumer configuration override
urbandan commented on a change in pull request #9430:
URL: https://github.com/apache/kafka/pull/9430#discussion_r569494586
##########
File path: core/src/main/scala/kafka/tools/GetOffsetShell.scala
##########
@@ -33,49 +34,65 @@ 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 topicOpt = parser.accepts("topic", "REQUIRED: The topic to get offset from.")
+ 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 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])
+ 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", "comma separated list of partition ids. If not specified, it will find offsets for all partitions")
+ 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)
- parser.accepts("offsets", "DEPRECATED AND IGNORED: number of offsets returned")
- .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.")
+ val commandConfigOpt = parser.accepts("command-config", s"Property file containing configs to be passed to Consumer Client.")
.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)
+ 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 topic = options.valueOf(topicOpt)
+ 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] = {
Review comment:
Good catch, thank you
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org