Onur Karaman
10 years ago
committed by
Neha Narkhede
6 changed files with 641 additions and 48 deletions
@ -0,0 +1,310 @@
@@ -0,0 +1,310 @@
|
||||
/** |
||||
* 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.admin |
||||
|
||||
|
||||
import kafka.utils._ |
||||
import org.I0Itec.zkclient.ZkClient |
||||
import kafka.common._ |
||||
import java.util.Properties |
||||
import kafka.client.ClientUtils |
||||
import kafka.api.{OffsetRequest, PartitionOffsetRequestInfo, OffsetFetchResponse, OffsetFetchRequest} |
||||
import org.I0Itec.zkclient.exception.ZkNoNodeException |
||||
import kafka.common.TopicAndPartition |
||||
import joptsimple.{OptionSpec, OptionParser} |
||||
import scala.collection.{Set, mutable} |
||||
import kafka.consumer.SimpleConsumer |
||||
import collection.JavaConversions._ |
||||
|
||||
|
||||
object ConsumerGroupCommand { |
||||
|
||||
def main(args: Array[String]) { |
||||
val opts = new ConsumerGroupCommandOptions(args) |
||||
|
||||
if(args.length == 0) |
||||
CommandLineUtils.printUsageAndDie(opts.parser, "List all consumer groups, describe a consumer group, or delete consumer group info.") |
||||
|
||||
// should have exactly one action |
||||
val actions = Seq(opts.listOpt, opts.describeOpt, opts.deleteOpt).count(opts.options.has _) |
||||
if(actions != 1) |
||||
CommandLineUtils.printUsageAndDie(opts.parser, "Command must include exactly one action: --list, --describe, --delete") |
||||
|
||||
opts.checkArgs() |
||||
|
||||
val zkClient = new ZkClient(opts.options.valueOf(opts.zkConnectOpt), 30000, 30000, ZKStringSerializer) |
||||
|
||||
try { |
||||
if (opts.options.has(opts.listOpt)) |
||||
list(zkClient) |
||||
else if (opts.options.has(opts.describeOpt)) |
||||
describe(zkClient, opts) |
||||
else if (opts.options.has(opts.deleteOpt)) |
||||
delete(zkClient, opts) |
||||
} catch { |
||||
case e: Throwable => |
||||
println("Error while executing consumer group command " + e.getMessage) |
||||
println(Utils.stackTrace(e)) |
||||
} finally { |
||||
zkClient.close() |
||||
} |
||||
} |
||||
|
||||
def list(zkClient: ZkClient) { |
||||
ZkUtils.getConsumerGroups(zkClient).foreach(println) |
||||
} |
||||
|
||||
def describe(zkClient: ZkClient, opts: ConsumerGroupCommandOptions) { |
||||
val configs = parseConfigs(opts) |
||||
val channelSocketTimeoutMs = configs.getProperty("channelSocketTimeoutMs", "600").toInt |
||||
val channelRetryBackoffMs = configs.getProperty("channelRetryBackoffMsOpt", "300").toInt |
||||
val group = opts.options.valueOf(opts.groupOpt) |
||||
val topics = ZkUtils.getTopicsByConsumerGroup(zkClient, group) |
||||
if (topics.isEmpty) { |
||||
println("No topic available for consumer group provided") |
||||
} |
||||
topics.foreach(topic => describeTopic(zkClient, group, topic, channelSocketTimeoutMs, channelRetryBackoffMs)) |
||||
} |
||||
|
||||
def delete(zkClient: ZkClient, opts: ConsumerGroupCommandOptions) { |
||||
if (opts.options.has(opts.groupOpt) && opts.options.has(opts.topicOpt)) { |
||||
deleteForTopic(zkClient, opts) |
||||
} |
||||
else if (opts.options.has(opts.groupOpt)) { |
||||
deleteForGroup(zkClient, opts) |
||||
} |
||||
else if (opts.options.has(opts.topicOpt)) { |
||||
deleteAllForTopic(zkClient, opts) |
||||
} |
||||
} |
||||
|
||||
private def deleteForGroup(zkClient: ZkClient, opts: ConsumerGroupCommandOptions) { |
||||
val groups = opts.options.valuesOf(opts.groupOpt) |
||||
groups.foreach { group => |
||||
try { |
||||
if (AdminUtils.deleteConsumerGroupInZK(zkClient, group)) |
||||
println("Deleted all consumer group information for group %s in zookeeper.".format(group)) |
||||
else |
||||
println("Delete for group %s failed because its consumers are still active.".format(group)) |
||||
} |
||||
catch { |
||||
case e: ZkNoNodeException => |
||||
println("Delete for group %s failed because group does not exist.".format(group)) |
||||
} |
||||
} |
||||
} |
||||
|
||||
private def deleteForTopic(zkClient: ZkClient, opts: ConsumerGroupCommandOptions) { |
||||
val groups = opts.options.valuesOf(opts.groupOpt) |
||||
val topic = opts.options.valueOf(opts.topicOpt) |
||||
Topic.validate(topic) |
||||
groups.foreach { group => |
||||
try { |
||||
if (AdminUtils.deleteConsumerGroupInfoForTopicInZK(zkClient, group, topic)) |
||||
println("Deleted consumer group information for group %s topic %s in zookeeper.".format(group, topic)) |
||||
else |
||||
println("Delete for group %s topic %s failed because its consumers are still active.".format(group, topic)) |
||||
} |
||||
catch { |
||||
case e: ZkNoNodeException => |
||||
println("Delete for group %s topic %s failed because group does not exist.".format(group, topic)) |
||||
} |
||||
} |
||||
} |
||||
|
||||
private def deleteAllForTopic(zkClient: ZkClient, opts: ConsumerGroupCommandOptions) { |
||||
val topic = opts.options.valueOf(opts.topicOpt) |
||||
Topic.validate(topic) |
||||
AdminUtils.deleteAllConsumerGroupInfoForTopicInZK(zkClient, topic) |
||||
println("Deleted consumer group information for all inactive consumer groups for topic %s in zookeeper.".format(topic)) |
||||
} |
||||
|
||||
private def parseConfigs(opts: ConsumerGroupCommandOptions): Properties = { |
||||
val configsToBeAdded = opts.options.valuesOf(opts.configOpt).map(_.split("""\s*=\s*""")) |
||||
require(configsToBeAdded.forall(config => config.length == 2), |
||||
"Invalid config: all configs to be added must be in the format \"key=val\".") |
||||
val props = new Properties |
||||
configsToBeAdded.foreach(pair => props.setProperty(pair(0).trim, pair(1).trim)) |
||||
props |
||||
} |
||||
|
||||
private def describeTopic(zkClient: ZkClient, |
||||
group: String, |
||||
topic: String, |
||||
channelSocketTimeoutMs: Int, |
||||
channelRetryBackoffMs: Int) { |
||||
val topicPartitions = getTopicPartitions(zkClient, topic) |
||||
val partitionOffsets = getPartitionOffsets(zkClient, group, topicPartitions, channelSocketTimeoutMs, channelRetryBackoffMs) |
||||
println("%s, %s, %s, %s, %s, %s, %s" |
||||
.format("GROUP", "TOPIC", "PARTITION", "CURRENT OFFSET", "LOG END OFFSET", "LAG", "OWNER")) |
||||
topicPartitions |
||||
.sortBy { case topicPartition => topicPartition.partition } |
||||
.foreach { topicPartition => |
||||
describePartition(zkClient, group, topicPartition.topic, topicPartition.partition, partitionOffsets.get(topicPartition)) |
||||
} |
||||
} |
||||
|
||||
private def getTopicPartitions(zkClient: ZkClient, topic: String) = { |
||||
val topicPartitionMap = ZkUtils.getPartitionsForTopics(zkClient, Seq(topic)) |
||||
val partitions = topicPartitionMap.getOrElse(topic, Seq.empty) |
||||
partitions.map(TopicAndPartition(topic, _)) |
||||
} |
||||
|
||||
private def getPartitionOffsets(zkClient: ZkClient, |
||||
group: String, |
||||
topicPartitions: Seq[TopicAndPartition], |
||||
channelSocketTimeoutMs: Int, |
||||
channelRetryBackoffMs: Int): Map[TopicAndPartition, Long] = { |
||||
val offsetMap = mutable.Map[TopicAndPartition, Long]() |
||||
val channel = ClientUtils.channelToOffsetManager(group, zkClient, channelSocketTimeoutMs, channelRetryBackoffMs) |
||||
channel.send(OffsetFetchRequest(group, topicPartitions)) |
||||
val offsetFetchResponse = OffsetFetchResponse.readFrom(channel.receive().buffer) |
||||
|
||||
offsetFetchResponse.requestInfo.foreach { case (topicAndPartition, offsetAndMetadata) => |
||||
if (offsetAndMetadata == OffsetMetadataAndError.NoOffset) { |
||||
val topicDirs = new ZKGroupTopicDirs(group, topicAndPartition.topic) |
||||
// this group may not have migrated off zookeeper for offsets storage (we don't expose the dual-commit option in this tool |
||||
// (meaning the lag may be off until all the consumers in the group have the same setting for offsets storage) |
||||
try { |
||||
val offset = ZkUtils.readData(zkClient, topicDirs.consumerOffsetDir + "/" + topicAndPartition.partition)._1.toLong |
||||
offsetMap.put(topicAndPartition, offset) |
||||
} catch { |
||||
case z: ZkNoNodeException => |
||||
println("Could not fetch offset from zookeeper for group %s partition %s due to missing offset data in zookeeper." |
||||
.format(group, topicAndPartition)) |
||||
} |
||||
} |
||||
else if (offsetAndMetadata.error == ErrorMapping.NoError) |
||||
offsetMap.put(topicAndPartition, offsetAndMetadata.offset) |
||||
else |
||||
println("Could not fetch offset from kafka for group %s partition %s due to %s." |
||||
.format(group, topicAndPartition, ErrorMapping.exceptionFor(offsetAndMetadata.error))) |
||||
} |
||||
channel.disconnect() |
||||
offsetMap.toMap |
||||
} |
||||
|
||||
private def describePartition(zkClient: ZkClient, |
||||
group: String, |
||||
topic: String, |
||||
partition: Int, |
||||
offsetOpt: Option[Long]) { |
||||
val topicAndPartition = TopicAndPartition(topic, partition) |
||||
val groupDirs = new ZKGroupTopicDirs(group, topic) |
||||
val owner = ZkUtils.readDataMaybeNull(zkClient, groupDirs.consumerOwnerDir + "/" + partition)._1 |
||||
ZkUtils.getLeaderForPartition(zkClient, topic, partition) match { |
||||
case Some(-1) => |
||||
println("%s, %s, %s, %s, %s, %s, %s" |
||||
.format(group, topic, partition, offsetOpt.getOrElse("unknown"), "unknown", "unknown", owner.getOrElse("none"))) |
||||
case Some(brokerId) => |
||||
val consumerOpt = getConsumer(zkClient, brokerId) |
||||
consumerOpt match { |
||||
case Some(consumer) => |
||||
val request = |
||||
OffsetRequest(Map(topicAndPartition -> PartitionOffsetRequestInfo(OffsetRequest.LatestTime, 1))) |
||||
val logEndOffset = consumer.getOffsetsBefore(request).partitionErrorAndOffsets(topicAndPartition).offsets.head |
||||
consumer.close() |
||||
|
||||
val lag = offsetOpt.filter(_ != -1).map(logEndOffset - _) |
||||
println("%s, %s, %s, %s, %s, %s, %s" |
||||
.format(group, topic, partition, offsetOpt.getOrElse("unknown"), logEndOffset, lag.getOrElse("unknown"), owner.getOrElse("none"))) |
||||
case None => // ignore |
||||
} |
||||
case None => |
||||
println("No broker for partition %s".format(topicAndPartition)) |
||||
} |
||||
} |
||||
|
||||
private def getConsumer(zkClient: ZkClient, brokerId: Int): Option[SimpleConsumer] = { |
||||
try { |
||||
ZkUtils.readDataMaybeNull(zkClient, ZkUtils.BrokerIdsPath + "/" + brokerId)._1 match { |
||||
case Some(brokerInfoString) => |
||||
Json.parseFull(brokerInfoString) match { |
||||
case Some(m) => |
||||
val brokerInfo = m.asInstanceOf[Map[String, Any]] |
||||
val host = brokerInfo.get("host").get.asInstanceOf[String] |
||||
val port = brokerInfo.get("port").get.asInstanceOf[Int] |
||||
Some(new SimpleConsumer(host, port, 10000, 100000, "ConsumerGroupCommand")) |
||||
case None => |
||||
throw new BrokerNotAvailableException("Broker id %d does not exist".format(brokerId)) |
||||
} |
||||
case None => |
||||
throw new BrokerNotAvailableException("Broker id %d does not exist".format(brokerId)) |
||||
} |
||||
} catch { |
||||
case t: Throwable => |
||||
println("Could not parse broker info due to " + t.getMessage) |
||||
None |
||||
} |
||||
} |
||||
|
||||
class ConsumerGroupCommandOptions(args: Array[String]) { |
||||
val ZkConnectDoc = "REQUIRED: The connection string for the zookeeper connection in the form host:port. " + |
||||
"Multiple URLS can be given to allow fail-over." |
||||
val GroupDoc = "The consumer group we wish to act on." |
||||
val TopicDoc = "The topic whose consumer group information should be deleted." |
||||
val ConfigDoc = "Configuration for timeouts. For instance --config channelSocketTimeoutMs=600" |
||||
val ListDoc = "List all consumer groups." |
||||
val DescribeDoc = "Describe consumer group and list offset lag related to given group." |
||||
val nl = System.getProperty("line.separator") |
||||
val DeleteDoc = "Pass in groups to delete topic partition offsets and ownership information " + |
||||
"over the entire consumer group. For instance --group g1 --group g2" + nl + |
||||
"Pass in groups with a single topic to just delete the given topic's partition offsets and ownership " + |
||||
"information for the given consumer groups. For instance --group g1 --group g2 --topic t1" + nl + |
||||
"Pass in just a topic to delete the given topic's partition offsets and ownership information " + |
||||
"for every consumer group. For instance --topic t1" + nl + |
||||
"WARNING: Only does deletions on consumer groups that are not active." |
||||
val parser = new OptionParser |
||||
val zkConnectOpt = parser.accepts("zookeeper", ZkConnectDoc) |
||||
.withRequiredArg |
||||
.describedAs("urls") |
||||
.ofType(classOf[String]) |
||||
val groupOpt = parser.accepts("group", GroupDoc) |
||||
.withRequiredArg |
||||
.describedAs("consumer group") |
||||
.ofType(classOf[String]) |
||||
val topicOpt = parser.accepts("topic", TopicDoc) |
||||
.withRequiredArg |
||||
.describedAs("topic") |
||||
.ofType(classOf[String]) |
||||
val configOpt = parser.accepts("config", ConfigDoc) |
||||
.withRequiredArg |
||||
.describedAs("name=value") |
||||
.ofType(classOf[String]) |
||||
val listOpt = parser.accepts("list", ListDoc) |
||||
val describeOpt = parser.accepts("describe", DescribeDoc) |
||||
val deleteOpt = parser.accepts("delete", DeleteDoc) |
||||
val options = parser.parse(args : _*) |
||||
|
||||
val allConsumerGroupLevelOpts: Set[OptionSpec[_]] = Set(listOpt, describeOpt, deleteOpt) |
||||
|
||||
def checkArgs() { |
||||
// check required args |
||||
CommandLineUtils.checkRequiredArgs(parser, options, zkConnectOpt) |
||||
if (options.has(describeOpt)) |
||||
CommandLineUtils.checkRequiredArgs(parser, options, groupOpt) |
||||
if (options.has(deleteOpt) && !options.has(groupOpt) && !options.has(topicOpt)) |
||||
CommandLineUtils.printUsageAndDie(parser, "Option %s either takes %s, %s, or both".format(deleteOpt, groupOpt, topicOpt)) |
||||
|
||||
// check invalid args |
||||
CommandLineUtils.checkInvalidArgs(parser, options, groupOpt, allConsumerGroupLevelOpts - describeOpt - deleteOpt) |
||||
CommandLineUtils.checkInvalidArgs(parser, options, topicOpt, allConsumerGroupLevelOpts - deleteOpt) |
||||
CommandLineUtils.checkInvalidArgs(parser, options, configOpt, allConsumerGroupLevelOpts - describeOpt) |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,212 @@
@@ -0,0 +1,212 @@
|
||||
/** |
||||
* 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.admin |
||||
|
||||
import org.scalatest.junit.JUnit3Suite |
||||
import kafka.utils.{ZKGroupDirs, ZKGroupTopicDirs, ZkUtils, TestUtils} |
||||
import kafka.server.KafkaConfig |
||||
import org.junit.Test |
||||
import kafka.consumer._ |
||||
import org.apache.kafka.clients.producer.{ProducerRecord, KafkaProducer} |
||||
import kafka.integration.KafkaServerTestHarness |
||||
|
||||
|
||||
class DeleteConsumerGroupTest extends JUnit3Suite with KafkaServerTestHarness { |
||||
val configs = TestUtils.createBrokerConfigs(3, false, true).map(new KafkaConfig(_)) |
||||
|
||||
@Test |
||||
def testGroupWideDeleteInZK() { |
||||
val topic = "test" |
||||
val groupToDelete = "groupToDelete" |
||||
val otherGroup = "otherGroup" |
||||
|
||||
TestUtils.createTopic(zkClient, topic, 1, 3, servers) |
||||
fillInConsumerGroupInfo(topic, groupToDelete, "consumer", 0, 10, false) |
||||
fillInConsumerGroupInfo(topic, otherGroup, "consumer", 0, 10, false) |
||||
|
||||
AdminUtils.deleteConsumerGroupInZK(zkClient, groupToDelete) |
||||
|
||||
TestUtils.waitUntilTrue(() => !groupDirExists(new ZKGroupDirs(groupToDelete)), |
||||
"DeleteConsumerGroupInZK should delete the provided consumer group's directory") |
||||
TestUtils.waitUntilTrue(() => groupDirExists(new ZKGroupDirs(otherGroup)), |
||||
"DeleteConsumerGroupInZK should not delete unrelated consumer group directories") |
||||
} |
||||
|
||||
@Test |
||||
def testGroupWideDeleteInZKDoesNothingForActiveConsumerGroup() { |
||||
val topic = "test" |
||||
val groupToDelete = "groupToDelete" |
||||
val otherGroup = "otherGroup" |
||||
|
||||
TestUtils.createTopic(zkClient, topic, 1, 3, servers) |
||||
fillInConsumerGroupInfo(topic, groupToDelete, "consumer", 0, 10, true) |
||||
fillInConsumerGroupInfo(topic, otherGroup, "consumer", 0, 10, false) |
||||
|
||||
AdminUtils.deleteConsumerGroupInZK(zkClient, groupToDelete) |
||||
|
||||
TestUtils.waitUntilTrue(() => groupDirExists(new ZKGroupDirs(groupToDelete)), |
||||
"DeleteConsumerGroupInZK should not delete the provided consumer group's directory if the consumer group is still active") |
||||
TestUtils.waitUntilTrue(() => groupDirExists(new ZKGroupDirs(otherGroup)), |
||||
"DeleteConsumerGroupInZK should not delete unrelated consumer group directories") |
||||
} |
||||
|
||||
@Test |
||||
def testGroupTopicWideDeleteInZKForGroupConsumingOneTopic() { |
||||
val topic = "test" |
||||
val groupToDelete = "groupToDelete" |
||||
val otherGroup = "otherGroup" |
||||
TestUtils.createTopic(zkClient, topic, 1, 3, servers) |
||||
fillInConsumerGroupInfo(topic, groupToDelete, "consumer", 0, 10, false) |
||||
fillInConsumerGroupInfo(topic, otherGroup, "consumer", 0, 10, false) |
||||
|
||||
AdminUtils.deleteConsumerGroupInfoForTopicInZK(zkClient, groupToDelete, topic) |
||||
|
||||
TestUtils.waitUntilTrue(() => !groupDirExists(new ZKGroupDirs(groupToDelete)), |
||||
"DeleteConsumerGroupInfoForTopicInZK should delete the provided consumer group's directory if it just consumes from one topic") |
||||
TestUtils.waitUntilTrue(() => groupTopicOffsetAndOwnerDirsExist(new ZKGroupTopicDirs(otherGroup, topic)), |
||||
"DeleteConsumerGroupInfoForTopicInZK should not delete unrelated consumer group owner and offset directories") |
||||
} |
||||
|
||||
@Test |
||||
def testGroupTopicWideDeleteInZKForGroupConsumingMultipleTopics() { |
||||
val topicToDelete = "topicToDelete" |
||||
val otherTopic = "otherTopic" |
||||
val groupToDelete = "groupToDelete" |
||||
val otherGroup = "otherGroup" |
||||
TestUtils.createTopic(zkClient, topicToDelete, 1, 3, servers) |
||||
TestUtils.createTopic(zkClient, otherTopic, 1, 3, servers) |
||||
|
||||
fillInConsumerGroupInfo(topicToDelete, groupToDelete, "consumer", 0, 10, false) |
||||
fillInConsumerGroupInfo(otherTopic, groupToDelete, "consumer", 0, 10, false) |
||||
fillInConsumerGroupInfo(topicToDelete, otherGroup, "consumer", 0, 10, false) |
||||
|
||||
AdminUtils.deleteConsumerGroupInfoForTopicInZK(zkClient, groupToDelete, topicToDelete) |
||||
|
||||
TestUtils.waitUntilTrue(() => !groupTopicOffsetAndOwnerDirsExist(new ZKGroupTopicDirs(groupToDelete, topicToDelete)), |
||||
"DeleteConsumerGroupInfoForTopicInZK should delete the provided consumer group's owner and offset directories for the given topic") |
||||
TestUtils.waitUntilTrue(() => groupTopicOffsetAndOwnerDirsExist(new ZKGroupTopicDirs(groupToDelete, otherTopic)), |
||||
"DeleteConsumerGroupInfoForTopicInZK should not delete the provided consumer group's owner and offset directories for unrelated topics") |
||||
TestUtils.waitUntilTrue(() => groupTopicOffsetAndOwnerDirsExist(new ZKGroupTopicDirs(otherGroup, topicToDelete)), |
||||
"DeleteConsumerGroupInfoForTopicInZK should not delete unrelated consumer group owner and offset directories") |
||||
} |
||||
|
||||
@Test |
||||
def testGroupTopicWideDeleteInZKDoesNothingForActiveGroupConsumingMultipleTopics() { |
||||
val topicToDelete = "topicToDelete" |
||||
val otherTopic = "otherTopic" |
||||
val group = "group" |
||||
TestUtils.createTopic(zkClient, topicToDelete, 1, 3, servers) |
||||
TestUtils.createTopic(zkClient, otherTopic, 1, 3, servers) |
||||
|
||||
fillInConsumerGroupInfo(topicToDelete, group, "consumer", 0, 10, true) |
||||
fillInConsumerGroupInfo(otherTopic, group, "consumer", 0, 10, true) |
||||
|
||||
AdminUtils.deleteConsumerGroupInfoForTopicInZK(zkClient, group, topicToDelete) |
||||
|
||||
TestUtils.waitUntilTrue(() => groupTopicOffsetAndOwnerDirsExist(new ZKGroupTopicDirs(group, topicToDelete)), |
||||
"DeleteConsumerGroupInfoForTopicInZK should not delete the provided consumer group's owner and offset directories for the given topic if the consumer group is still active") |
||||
TestUtils.waitUntilTrue(() => groupTopicOffsetAndOwnerDirsExist(new ZKGroupTopicDirs(group, otherTopic)), |
||||
"DeleteConsumerGroupInfoForTopicInZK should not delete the provided consumer group's owner and offset directories for unrelated topics") |
||||
} |
||||
|
||||
@Test |
||||
def testTopicWideDeleteInZK() { |
||||
val topicToDelete = "topicToDelete" |
||||
val otherTopic = "otherTopic" |
||||
val groups = Seq("group1", "group2") |
||||
|
||||
TestUtils.createTopic(zkClient, topicToDelete, 1, 3, servers) |
||||
TestUtils.createTopic(zkClient, otherTopic, 1, 3, servers) |
||||
val groupTopicDirsForTopicToDelete = groups.map(group => new ZKGroupTopicDirs(group, topicToDelete)) |
||||
val groupTopicDirsForOtherTopic = groups.map(group => new ZKGroupTopicDirs(group, otherTopic)) |
||||
groupTopicDirsForTopicToDelete.foreach(dir => fillInConsumerGroupInfo(topicToDelete, dir.group, "consumer", 0, 10, false)) |
||||
groupTopicDirsForOtherTopic.foreach(dir => fillInConsumerGroupInfo(otherTopic, dir.group, "consumer", 0, 10, false)) |
||||
|
||||
AdminUtils.deleteAllConsumerGroupInfoForTopicInZK(zkClient, topicToDelete) |
||||
|
||||
TestUtils.waitUntilTrue(() => !groupTopicDirsForTopicToDelete.exists(groupTopicOffsetAndOwnerDirsExist), |
||||
"Consumer group info on deleted topic topic should be deleted by DeleteAllConsumerGroupInfoForTopicInZK") |
||||
TestUtils.waitUntilTrue(() => groupTopicDirsForOtherTopic.forall(groupTopicOffsetAndOwnerDirsExist), |
||||
"Consumer group info on unrelated topics should not be deleted by DeleteAllConsumerGroupInfoForTopicInZK") |
||||
} |
||||
|
||||
@Test |
||||
def testConsumptionOnRecreatedTopicAfterTopicWideDeleteInZK() { |
||||
val topic = "topic" |
||||
val group = "group" |
||||
|
||||
TestUtils.createTopic(zkClient, topic, 1, 3, servers) |
||||
val dir = new ZKGroupTopicDirs(group, topic) |
||||
fillInConsumerGroupInfo(topic, dir.group, "consumer", 0, 10, false) |
||||
|
||||
AdminUtils.deleteTopic(zkClient, topic) |
||||
TestUtils.verifyTopicDeletion(zkClient, topic, 1, servers) |
||||
AdminUtils.deleteAllConsumerGroupInfoForTopicInZK(zkClient, topic) |
||||
|
||||
TestUtils.waitUntilTrue(() => !groupDirExists(dir), |
||||
"Consumer group info on related topics should be deleted by DeleteAllConsumerGroupInfoForTopicInZK") |
||||
//produce events |
||||
val producer = TestUtils.createNewProducer(brokerList) |
||||
produceEvents(producer, topic, List.fill(10)("test")) |
||||
|
||||
//consume events |
||||
val consumerProps = TestUtils.createConsumerProperties(zkConnect, group, "consumer") |
||||
consumerProps.put("auto.commit.enable", "false") |
||||
consumerProps.put("auto.offset.reset", "smallest") |
||||
consumerProps.put("consumer.timeout.ms", "2000") |
||||
consumerProps.put("fetch.wait.max.ms", "0") |
||||
val consumerConfig = new ConsumerConfig(consumerProps) |
||||
val consumerConnector: ConsumerConnector = Consumer.create(consumerConfig) |
||||
val messageStream = consumerConnector.createMessageStreams(Map(topic -> 1))(topic).head |
||||
consumeEvents(messageStream, 5) |
||||
consumerConnector.commitOffsets(false) |
||||
producer.close() |
||||
consumerConnector.shutdown() |
||||
|
||||
TestUtils.waitUntilTrue(() => groupTopicOffsetAndOwnerDirsExist(dir), |
||||
"Consumer group info should exist after consuming from a recreated topic") |
||||
} |
||||
|
||||
private def fillInConsumerGroupInfo(topic: String, group: String, consumerId: String, partition: Int, offset: Int, registerConsumer: Boolean) { |
||||
val consumerProps = TestUtils.createConsumerProperties(zkConnect, group, consumerId) |
||||
val consumerConfig = new ConsumerConfig(consumerProps) |
||||
val dir = new ZKGroupTopicDirs(group, topic) |
||||
TestUtils.updateConsumerOffset(consumerConfig, dir.consumerOffsetDir + "/" + partition, offset) |
||||
ZkUtils.createEphemeralPathExpectConflict(zkClient, ZkUtils.getConsumerPartitionOwnerPath(group, topic, partition), "") |
||||
ZkUtils.makeSurePersistentPathExists(zkClient, dir.consumerRegistryDir) |
||||
if (registerConsumer) { |
||||
ZkUtils.createEphemeralPathExpectConflict(zkClient, dir.consumerRegistryDir + "/" + consumerId, "") |
||||
} |
||||
} |
||||
|
||||
private def groupDirExists(dir: ZKGroupDirs) = { |
||||
ZkUtils.pathExists(zkClient, dir.consumerGroupDir) |
||||
} |
||||
|
||||
private def groupTopicOffsetAndOwnerDirsExist(dir: ZKGroupTopicDirs) = { |
||||
ZkUtils.pathExists(zkClient, dir.consumerOffsetDir) && ZkUtils.pathExists(zkClient, dir.consumerOwnerDir) |
||||
} |
||||
|
||||
private def produceEvents(producer: KafkaProducer[Array[Byte], Array[Byte]], topic: String, messages: List[String]) { |
||||
messages.foreach(message => producer.send(new ProducerRecord(topic, message.getBytes))) |
||||
} |
||||
|
||||
private def consumeEvents(messageStream: KafkaStream[Array[Byte], Array[Byte]], n: Int) { |
||||
val iter = messageStream.iterator |
||||
(0 until n).foreach(_ => iter.next) |
||||
} |
||||
} |
||||
Loading…
Reference in new issue