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/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 7606247ebd6c1..dfd5a227689ec 100644 --- a/core/src/main/scala/kafka/tools/GetOffsetShell.scala +++ b/core/src/main/scala/kafka/tools/GetOffsetShell.scala @@ -20,135 +20,213 @@ package kafka.tools import java.util.Properties import joptsimple._ -import kafka.utils.{CommandLineUtils, Exit, ToolsUtils} +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]*)-([0-9]*))))?") def main(args: Array[String]): Unit = { + try { + fetchOffsets(args) + } catch { + case e: Exception => + println(s"Error occurred: ${e.getMessage}") + Exit.exit(1, Some(e.getMessage)) + } + } + + private def fetchOffsets(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", 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("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 .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 partitionIdsRequested: Set[Int] = { - val partitionsString = options.valueOf(partitionOpt) - if (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 excludeInternalTopics = options.has(excludeInternalTopicsOpt) + + if (options.has(topicPartitionsOpt) && (options.has(topicOpt) || options.has(partitionsOpt))) { + throw new IllegalArgumentException("--topic-partitions cannot be used with --topic or --partitions") } + val listOffsetsTimestamp = options.valueOf(timeOpt).longValue - val config = new Properties + val topicPartitionFilter = if (options.has(topicPartitionsOpt)) { + createTopicPartitionFilterWithPatternList(options.valueOf(topicPartitionsOpt), excludeInternalTopics) + } else { + val partitionIdsRequested = createPartitionSet(options.valueOf(partitionsOpt)) + + 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 - } + try { + 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) { + throw new IllegalArgumentException("Could not match any topic-partitions with the specified filters") } - } - val topicPartitions = partitionInfos.sortBy(_.partition).flatMap { p => - if (p.leader == null) { - System.err.println(s"Error: partition ${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) + } + } + + 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() } + } - /* 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) - } + 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: + * List: TopicPartitionPattern(, TopicPartitionPattern)* + * TopicPartitionPattern: TopicPattern(:PartitionPattern)? | :PartitionPattern + * TopicPattern: REGEX + * PartitionPattern: NUMBER | NUMBER-(NUMBER)? | -NUMBER + */ + 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) } + } + + def parseRuleSpec(ruleSpec: String, excludeInternalTopics: Boolean): PartitionInfo => Boolean = { + val matcher = TopicPartitionPattern.matcher(ruleSpec) + if (!matcher.matches()) + throw new IllegalArgumentException(s"Invalid rule specification: $ruleSpec") + + def group(group: Int): Option[String] = { + Option(matcher.group(group)).filter(s => s != null && s.nonEmpty) } - partitionOffsets.toArray.sortBy { case (tp, _) => tp.partition }.foreach { case (tp, offset) => - println(s"$topic:${tp.partition}:${Option(offset).getOrElse("")}") + 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 => topicFilter.isTopicAllowed(tp.topic, excludeInternalTopics) && partitionFilter(tp.partition) } /** - * Return the partition infos for `topic`. If the topic does not exist, `None` is returned. + * Creates a topic-partition filter based on a topic pattern and a set of partition ids. */ - 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) + 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)) + } + + def createPartitionSet(partitionsString: String): Set[Int] = { + if (partitionsString == null || partitionsString.isEmpty) + Set.empty else - Some(partitionInfos.filter(p => partitionIds.contains(p.partition))) + 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(topicPartitionFilter) + }.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..edfadea401eca --- /dev/null +++ b/core/src/test/scala/kafka/tools/GetOffsetShellParsingTest.scala @@ -0,0 +1,207 @@ +/** + * 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.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(partitionInfo("test", 0))) + assertTrue(filter.apply(partitionInfo("test", 1))) + assertFalse(filter.apply(partitionInfo("test1", 0))) + assertFalse(filter.apply(partitionInfo("__consumer_offsets", 0))) + } + + @ParameterizedTest + @ValueSource(booleans = Array(true, false)) + def testTopicPartitionFilterForInternalTopicName(excludeInternal: Boolean): Unit = { + val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList("__consumer_offsets", excludeInternal) + 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))) + } + + @ParameterizedTest + @ValueSource(booleans = Array(true, false)) + def testTopicPartitionFilterForTopicNameList(excludeInternal: Boolean): Unit = { + val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList("test,test1,__consumer_offsets", excludeInternal) + assertTrue(filter.apply(partitionInfo("test", 0))) + assertTrue(filter.apply(partitionInfo("test1", 1))) + assertFalse(filter.apply(partitionInfo("test2", 0))) + + assertEquals(!excludeInternal, filter.apply(partitionInfo("__consumer_offsets", 0))) + } + + @ParameterizedTest + @ValueSource(booleans = Array(true, false)) + def testTopicPartitionFilterForRegex(excludeInternal: Boolean): Unit = { + val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList("test.*", excludeInternal) + 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))) + } + + @ParameterizedTest + @ValueSource(booleans = Array(true, false)) + def testTopicPartitionFilterForPartitionIndexSpec(excludeInternal: Boolean): Unit = { + val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList(":0", excludeInternal) + assertTrue(filter.apply(partitionInfo("test", 0))) + assertTrue(filter.apply(partitionInfo("test1", 0))) + assertFalse(filter.apply(partitionInfo("test2", 1))) + + assertEquals(!excludeInternal, filter.apply(partitionInfo("__consumer_offsets", 0))) + assertFalse(filter.apply(partitionInfo("__consumer_offsets", 1))) + } + + @ParameterizedTest + @ValueSource(booleans = Array(true, false)) + def testTopicPartitionFilterForPartitionRangeSpec(excludeInternal: Boolean): Unit = { + val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList(":1-3", excludeInternal) + 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(partitionInfo("__consumer_offsets", 2))) + assertFalse(filter.apply(partitionInfo("__consumer_offsets", 3))) + } + + @ParameterizedTest + @ValueSource(booleans = Array(true, false)) + def testTopicPartitionFilterForPartitionLowerBoundSpec(excludeInternal: Boolean): Unit = { + val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList(":1-", excludeInternal) + assertTrue(filter.apply(partitionInfo("test", 1))) + assertTrue(filter.apply(partitionInfo("test1", 2))) + assertFalse(filter.apply(partitionInfo("test2", 0))) + + assertEquals(!excludeInternal, filter.apply(partitionInfo("__consumer_offsets", 2))) + assertFalse(filter.apply(partitionInfo("__consumer_offsets", 0))) + } + + @ParameterizedTest + @ValueSource(booleans = Array(true, false)) + def testTopicPartitionFilterForPartitionUpperBoundSpec(excludeInternal: Boolean): Unit = { + val filter = GetOffsetShell.createTopicPartitionFilterWithPatternList(":-3", excludeInternal) + 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(partitionInfo("__consumer_offsets", 2))) + assertFalse(filter.apply(partitionInfo("__consumer_offsets", 3))) + } + + @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(partitionInfo("test", 0))) + assertTrue(filter.apply(partitionInfo("test", 3))) + assertFalse(filter.apply(partitionInfo("test", 1))) + + assertTrue(filter.apply(partitionInfo("test1", 0))) + assertTrue(filter.apply(partitionInfo("test1", 3))) + assertFalse(filter.apply(partitionInfo("test1", 1))) + + assertTrue(filter.apply(partitionInfo("custom", 3))) + assertFalse(filter.apply(partitionInfo("custom", 0))) + + 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.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.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.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.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 + def testPartitionFilterForInvalidSingleIndex(): Unit = { + assertThrows(classOf[IllegalArgumentException], + () => GetOffsetShell.createTopicPartitionFilterWithPatternList(":a", excludeInternalTopics = false)) + } + + @Test + def testPartitionFilterForInvalidRange(): Unit = { + assertThrows(classOf[IllegalArgumentException], + () => GetOffsetShell.createTopicPartitionFilterWithPatternList(":a-b", excludeInternalTopics = false)) + } + + @Test + def testPartitionFilterForInvalidLowerBound(): Unit = { + assertThrows(classOf[IllegalArgumentException], + () => GetOffsetShell.createTopicPartitionFilterWithPatternList(":a-", excludeInternalTopics = false)) + } + + @Test + def testPartitionFilterForInvalidUpperBound(): Unit = { + assertThrows(classOf[IllegalArgumentException], + () => GetOffsetShell.createTopicPartitionFilterWithPatternList(":-b", excludeInternalTopics = 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 new file mode 100644 index 0000000000000..cd611de177305 --- /dev/null +++ b/core/src/test/scala/kafka/tools/GetOffsetShellTest.scala @@ -0,0 +1,207 @@ +/** + * 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.util.Properties +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.producer.{KafkaProducer, ProducerConfig, ProducerRecord} +import org.apache.kafka.common.serialization.StringSerializer +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.{BeforeEach, Test} + +class GetOffsetShellTest extends KafkaServerTestHarness with Logging { + private val topicCount = 4 + private val offsetTopicPartitionCount = 4 + + override def generateConfigs: collection.Seq[KafkaConfig] = TestUtils.createBrokerConfigs(1, zkConnect) + .map { p => + p.put(KafkaConfig.OffsetsTopicPartitionsProp, Int.box(offsetTopicPartitionCount)) + p + }.map(KafkaConfig.fromProps) + + @BeforeEach + 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]) + + // 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() + + TestUtils.createOffsetsTopic(zkClient, servers) + } + + @Test + def testNoFilterOptions(): Unit = { + val offsets = executeAndParse(Array()) + assertEquals(expectedOffsetsWithInternal(), offsets) + } + + @Test + def testInternalExcluded(): Unit = { + val offsets = executeAndParse(Array("--exclude-internal-topics")) + assertEquals(expectedTestTopicOffsets(), offsets) + } + + @Test + def testTopicNameArg(): Unit = { + Range(1, topicCount + 1).foreach(i => { + val offsets = executeAndParse(Array("--topic", topicName(i))) + assertEquals(expectedOffsetsForTopic(i), offsets, () => "Offset output did not match for " + topicName(i)) + }) + } + + @Test + def testTopicPatternArg(): Unit = { + val offsets = executeAndParse(Array("--topic", "topic.*")) + assertEquals(expectedTestTopicOffsets(), offsets) + } + + @Test + def testPartitionsArg(): Unit = { + val offsets = executeAndParse(Array("--partitions", "0,1")) + assertEquals(expectedOffsetsWithInternal().filter { case (_, partition, _) => partition <= 1 }, offsets) + } + + @Test + def testTopicPatternArgWithPartitionsArg(): Unit = { + val offsets = executeAndParse(Array("--topic", "topic.*", "--partitions", "0,1")) + 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")) + assertEquals( + List( + ("__consumer_offsets", 3, Some(0)), + ("topic1", 0, Some(1)), + ("topic2", 1, Some(2)), + ("topic3", 2, Some(3)), + ("topic4", 2, Some(4)) + ), + offsets + ) + } + + @Test + def testTopicPartitionsArgWithInternalExcluded(): Unit = { + val offsets = executeAndParse(Array("--topic-partitions", + "topic1:0,topic2:1,topic(3|4):2,__.*:3", "--exclude-internal-topics")) + assertEquals( + List( + ("topic1", 0, Some(1)), + ("topic2", 1, Some(2)), + ("topic3", 2, Some(3)), + ("topic4", 2, Some(4)) + ), + offsets + ) + } + + @Test + def testTopicPartitionsNotFoundForNonExistentTopic(): Unit = { + assertExitCodeIsOne(Array("--topic", "some_nonexistent_topic")) + } + + @Test + def testTopicPartitionsNotFoundForExcludedInternalTopic(): Unit = { + assertExitCodeIsOne(Array("--topic", "some_nonexistent_topic:*")) + } + + @Test + def testTopicPartitionsNotFoundForNonMatchingTopicPartitionPattern(): Unit = { + assertExitCodeIsOne(Array("--topic-partitions", "__consumer_offsets", "--exclude-internal-topics")) + } + + @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 expectedTestTopicOffsets(): List[(String, Int, Option[Long])] = { + Range(1, topicCount + 1).flatMap(i => expectedOffsetsForTopic(i)).toList + } + + private def expectedOffsetsForTopic(i: Int): List[(String, Int, Option[Long])] = { + val name = topicName(i) + Range(0, i).map(p => (name, p, Some(i.toLong))).toList + } + + 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]): List[(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) + } + .toList + } + + 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) + } +} + + diff --git a/tests/kafkatest/services/kafka/kafka.py b/tests/kafkatest/services/kafka/kafka.py index af5a753f7e51c..4da2ab2e183d0 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 += " --bootstrap-server %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