Skip to content

KAFKA-12596: remove --zookeeper option from topic command #10457

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ijuma merged 9 commits into from
Jun 1, 2021
Merged

KAFKA-12596: remove --zookeeper option from topic command #10457

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
47f0058
KAFKA-12588: remove --zookeeper option from topic command
showuon Mar 31, 2021
4213cfc
KAFKA-12596: remove unused trait and refactor codes
showuon Apr 2, 2021
3b6cc27
KAFKA-12596: rename TopicCommandWithAdminClientTest and move to integ…
showuon Apr 3, 2021
9488dff
KAFKA-12596: move the integration test back to unit.kafka.admin
showuon Apr 3, 2021
79712cb
Merge branch 'trunk' of https://github.com/apache/kafka into KAFKA-12596
showuon Apr 4, 2021
71a2648
KAFKA-12596: check --bootstrap-server option manually
showuon Apr 7, 2021
a15b4a6
Merge branch 'trunk' of https://github.com/apache/kafka into KAFKA-12596
showuon Apr 16, 2021
a70a06b
Merge branch 'trunk' of https://github.com/apache/kafka into KAFKA-12596
showuon May 18, 2021
f22e330
KAFKA-12596: remove unused bootstrapOpt and add tests
showuon May 20, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
198 changes: 15 additions & 183 deletions core/src/main/scala/kafka/admin/TopicCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,23 +23,18 @@ import java.util.{Collections, Properties}
import joptsimple._
import kafka.common.AdminCommandFailedException
import kafka.log.LogConfig
import kafka.server.ConfigType
import kafka.utils.Implicits._
import kafka.utils._
import kafka.zk.{AdminZkClient, KafkaZkClient}
import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.clients.admin.CreatePartitionsOptions
import org.apache.kafka.clients.admin.CreateTopicsOptions
import org.apache.kafka.clients.admin.DeleteTopicsOptions
import org.apache.kafka.clients.admin.{Admin, ConfigEntry, ListTopicsOptions, NewPartitions, NewTopic, PartitionReassignment, Config => JConfig}
import org.apache.kafka.common.{Node, TopicPartition, TopicPartitionInfo, Uuid}
import org.apache.kafka.clients.admin.{Admin, ListTopicsOptions, NewPartitions, NewTopic, PartitionReassignment, Config => JConfig}
import org.apache.kafka.common.{TopicPartition, TopicPartitionInfo, Uuid}
import org.apache.kafka.common.config.ConfigResource.Type
import org.apache.kafka.common.config.{ConfigResource, TopicConfig}
import org.apache.kafka.common.errors.{ClusterAuthorizationException, InvalidTopicException, TopicExistsException, UnsupportedVersionException}
import org.apache.kafka.common.errors.{ClusterAuthorizationException, TopicExistsException, UnsupportedVersionException}
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.security.JaasUtils
import org.apache.kafka.common.utils.{Time, Utils}
import org.apache.zookeeper.KeeperException.NodeExistsException
import org.apache.kafka.common.utils.Utils

import scala.jdk.CollectionConverters._
import scala.collection._
Expand All @@ -51,11 +46,8 @@ object TopicCommand extends Logging {
def main(args: Array[String]): Unit = {
val opts = new TopicCommandOptions(args)
opts.checkArgs()

val topicService = if (opts.zkConnect.isDefined)
ZookeeperTopicService(opts.zkConnect)
else
AdminClientTopicService(opts.commandConfig, opts.bootstrapServer)

val topicService = AdminClientTopicService(opts.commandConfig, opts.bootstrapServer)

var exitCode = 0
try {
Expand Down Expand Up @@ -373,159 +365,6 @@ object TopicCommand extends Logging {
override def close(): Unit = adminClient.close()
}

object ZookeeperTopicService {
def apply(zkConnect: Option[String]): ZookeeperTopicService =
new ZookeeperTopicService(KafkaZkClient(zkConnect.get, JaasUtils.isZkSaslEnabled, 30000, 30000,
Int.MaxValue, Time.SYSTEM))
}

case class ZookeeperTopicService(zkClient: KafkaZkClient) extends TopicService {

override def createTopic(topic: CommandTopicPartition): Unit = {
val adminZkClient = new AdminZkClient(zkClient)
try {
if (topic.hasReplicaAssignment)
adminZkClient.createTopicWithAssignment(topic.name, topic.configsToAdd, topic.replicaAssignment.get)
else
adminZkClient.createTopic(topic.name, topic.partitions.get, topic.replicationFactor.get, topic.configsToAdd, topic.rackAwareMode)
println(s"Created topic ${topic.name}.")
} catch {
case e: TopicExistsException => if (!topic.ifTopicDoesntExist()) throw e
}
}

override def listTopics(opts: TopicCommandOptions): Unit = {
val topics = getTopics(opts.topic, opts.excludeInternalTopics)
for(topic <- topics) {
if (zkClient.isTopicMarkedForDeletion(topic))
println(s"$topic - marked for deletion")
else
println(topic)
}
}

override def alterTopic(opts: TopicCommandOptions): Unit = {
val topics = getTopics(opts.topic, opts.excludeInternalTopics)
val tp = new CommandTopicPartition(opts)
ensureTopicExists(topics, opts.topic, !opts.ifExists)
val adminZkClient = new AdminZkClient(zkClient)
topics.foreach { topic =>
val configs = adminZkClient.fetchEntityConfig(ConfigType.Topic, topic)
if(opts.topicConfig.isDefined || opts.configsToDelete.isDefined) {
println("WARNING: Altering topic configuration from this script has been deprecated and may be removed in future releases.")
println(" Going forward, please use kafka-configs.sh for this functionality")

// compile the final set of configs
configs ++= tp.configsToAdd
tp.configsToDelete.foreach(config => configs.remove(config))
adminZkClient.changeTopicConfig(topic, configs)
println(s"Updated config for topic $topic.")
}

if(tp.hasPartitions) {
if (Topic.isInternal(topic)) {
throw new IllegalArgumentException(s"The number of partitions for the internal topic $topic cannot be changed.")
}
println("WARNING: If partitions are increased for a topic that has a key, the partition " +
"logic or ordering of the messages will be affected")
val existingAssignment = zkClient.getFullReplicaAssignmentForTopics(immutable.Set(topic)).map {
case (topicPartition, assignment) => topicPartition.partition -> assignment
}
if (existingAssignment.isEmpty)
throw new InvalidTopicException(s"The topic $topic does not exist")
val newAssignment = tp.replicaAssignment.getOrElse(Map()).drop(existingAssignment.size)
val allBrokers = adminZkClient.getBrokerMetadatas()
val partitions: Integer = tp.partitions.getOrElse(1)
adminZkClient.addPartitions(topic, existingAssignment, allBrokers, partitions, Option(newAssignment).filter(_.nonEmpty))
println("Adding partitions succeeded!")
}
}
}

override def describeTopic(opts: TopicCommandOptions): Unit = {
val topics = getTopics(opts.topic, opts.excludeInternalTopics)
ensureTopicExists(topics, opts.topic, !opts.ifExists)
val liveBrokers = zkClient.getAllBrokersInCluster.map(broker => broker.id -> broker).toMap
val liveBrokerIds = liveBrokers.keySet
val describeOptions = new DescribeOptions(opts, liveBrokerIds)
val adminZkClient = new AdminZkClient(zkClient)

for (topic <- topics) {
zkClient.getReplicaAssignmentAndTopicIdForTopics(immutable.Set(topic)).headOption match {
case Some(replicaAssignmentAndTopicId) =>
val markedForDeletion = zkClient.isTopicMarkedForDeletion(topic)
if (describeOptions.describeConfigs) {
val configs = adminZkClient.fetchEntityConfig(ConfigType.Topic, topic).asScala
if (!opts.reportOverriddenConfigs || configs.nonEmpty) {
val numPartitions = replicaAssignmentAndTopicId.assignment.size
val replicationFactor = replicaAssignmentAndTopicId.assignment.head._2.replicas.size
val config = new JConfig(configs.map{ case (k, v) => new ConfigEntry(k, v) }.asJavaCollection)

val topicDesc = TopicDescription(topic,
replicaAssignmentAndTopicId.topicId.getOrElse(Uuid.ZERO_UUID), numPartitions, replicationFactor, config, markedForDeletion)
topicDesc.printDescription()
}
}
if (describeOptions.describePartitions) {
for ((tp, replicaAssignment) <- replicaAssignmentAndTopicId.assignment.toSeq.sortBy(_._1.partition())) {
val assignedReplicas = replicaAssignment.replicas
val (leaderOpt, isr) = zkClient.getTopicPartitionState(tp).map(_.leaderAndIsr) match {
case Some(leaderAndIsr) => (leaderAndIsr.leaderOpt, leaderAndIsr.isr)
case None => (None, Seq.empty[Int])
}

def asNode(brokerId: Int): Node = {
liveBrokers.get(brokerId) match {
case Some(broker) => broker.node(broker.endPoints.head.listenerName)
case None => new Node(brokerId, "", -1)
}
}

val info = new TopicPartitionInfo(tp.partition(), leaderOpt.map(asNode).orNull,
assignedReplicas.map(asNode).toList.asJava,
isr.map(asNode).toList.asJava)

val partitionDesc = PartitionDescription(topic, info, config = None, markedForDeletion, reassignment = None)
describeOptions.maybePrintPartitionDescription(partitionDesc)
}
}
case None =>
println("Topic " + topic + " doesn't exist!")
}
}
}

override def deleteTopic(opts: TopicCommandOptions): Unit = {
val topics = getTopics(opts.topic, opts.excludeInternalTopics)
ensureTopicExists(topics, opts.topic, !opts.ifExists)
topics.foreach { topic =>
try {
if (Topic.isInternal(topic)) {
throw new AdminOperationException(s"Topic $topic is a kafka internal topic and is not allowed to be marked for deletion.")
} else {
zkClient.createDeleteTopicPath(topic)
println(s"Topic $topic is marked for deletion.")
println("Note: This will have no impact if delete.topic.enable is not set to true.")
}
} catch {
case _: NodeExistsException =>
println(s"Topic $topic is already marked for deletion.")
case e: AdminOperationException =>
throw e
case e: Throwable =>
throw new AdminOperationException(s"Error while deleting topic $topic", e)
}
}
}

override def getTopics(topicIncludelist: Option[String], excludeInternalTopics: Boolean = false): Seq[String] = {
val allTopics = zkClient.getAllTopicsInCluster().toSeq.sorted
doGetTopics(allTopics, topicIncludelist, excludeInternalTopics)
}

override def close(): Unit = zkClient.close()
}

/**
* ensures topic existence and throws exception if topic doesn't exist
*
Expand Down Expand Up @@ -610,7 +449,7 @@ object TopicCommand extends Logging {
}

class TopicCommandOptions(args: Array[String]) extends CommandDefaultOptions(args) {
private val bootstrapServerOpt = parser.accepts("bootstrap-server", "REQUIRED: The Kafka server to connect to. In case of providing this, a direct Zookeeper connection won't be required.")
private val bootstrapServerOpt = parser.accepts("bootstrap-server", "REQUIRED: The Kafka server to connect to.")
.withRequiredArg
.describedAs("server to connect to")
.ofType(classOf[String])
Expand All @@ -619,11 +458,7 @@ object TopicCommand extends Logging {
.withRequiredArg
.describedAs("command config property file")
.ofType(classOf[String])
private val zkConnectOpt = parser.accepts("zookeeper", "DEPRECATED, The connection string for the zookeeper connection in the form host:port. " +
"Multiple hosts can be given to allow fail-over.")
.withRequiredArg
.describedAs("hosts")
.ofType(classOf[String])

private val listOpt = parser.accepts("list", "List all available topics.")
private val createOpt = parser.accepts("create", "Create a new topic.")
private val deleteOpt = parser.accepts("delete", "Delete a topic")
Expand Down Expand Up @@ -670,9 +505,9 @@ object TopicCommand extends Logging {
private val reportUnavailablePartitionsOpt = parser.accepts("unavailable-partitions",
"if set when describing topics, only show partitions whose leader is not available")
private val reportUnderMinIsrPartitionsOpt = parser.accepts("under-min-isr-partitions",
"if set when describing topics, only show partitions whose isr count is less than the configured minimum. Not supported with the --zookeeper option.")
"if set when describing topics, only show partitions whose isr count is less than the configured minimum.")
private val reportAtMinIsrPartitionsOpt = parser.accepts("at-min-isr-partitions",
"if set when describing topics, only show partitions whose isr count is equal to the configured minimum. Not supported with the --zookeeper option.")
"if set when describing topics, only show partitions whose isr count is equal to the configured minimum.")

@ijuma ijuma Apr 1, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We have this line:

    // This is not currently used, but we keep it for compatibility
    parser.accepts("force", "Suppress console prompts")

Was this deprecated also? Can we remove it?

@showuon showuon Apr 2, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I confirmed this issue is fixed in KIP-74 (KAFKA-2063). Yes I'll remove it!. Thank you.

private val topicsWithOverridesOpt = parser.accepts("topics-with-overrides",
"if set when describing topics, only show topics that have overridden configs")
private val ifExistsOpt = parser.accepts("if-exists",
Expand Down Expand Up @@ -704,7 +539,6 @@ object TopicCommand extends Logging {
def hasDescribeOption: Boolean = has(describeOpt)
def hasDeleteOption: Boolean = has(deleteOpt)

def zkConnect: Option[String] = valueAsOption(zkConnectOpt)
def bootstrapServer: Option[String] = valueAsOption(bootstrapServerOpt)
def commandConfig: Properties = if (has(commandConfigOpt)) Utils.loadProps(options.valueOf(commandConfigOpt)) else new Properties()
def topic: Option[String] = valueAsOption(topicOpt)
Expand Down Expand Up @@ -739,16 +573,14 @@ object TopicCommand extends Logging {
CommandLineUtils.printUsageAndDie(parser, "Command must include exactly one action: --list, --describe, --create, --alter or --delete")

// check required args
if (has(bootstrapServerOpt) == has(zkConnectOpt))
throw new IllegalArgumentException("Only one of --bootstrap-server or --zookeeper must be specified")

if (!has(bootstrapServerOpt))
CommandLineUtils.checkRequiredArgs(parser, options, zkConnectOpt)
throw new IllegalArgumentException("--bootstrap-server must be specified")

@ijuma ijuma Apr 1, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think you can mark the argument as required to get this behavior automatically.

@showuon showuon Apr 2, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good suggestion! Updated


if(has(describeOpt) && has(ifExistsOpt))
CommandLineUtils.checkRequiredArgs(parser, options, topicOpt)
if (!has(listOpt) && !has(describeOpt))
CommandLineUtils.checkRequiredArgs(parser, options, topicOpt)
if (has(createOpt) && !has(replicaAssignmentOpt) && has(zkConnectOpt))
if (has(createOpt) && !has(replicaAssignmentOpt))

@ijuma ijuma May 19, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Was it a bug that we only verified this when zkConnectOpt was set? If so, we should add a unit test for this case.

@showuon showuon May 20, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch! Added 2 tests: testCreateWithAssignmentAndPartitionCount and testCreateWithAssignmentAndReplicationFactor in TopicCommandTest. Thank you.

CommandLineUtils.checkRequiredArgs(parser, options, partitionsOpt, replicationFactorOpt)
if (has(bootstrapServerOpt) && has(alterOpt)) {

@ijuma ijuma Apr 1, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We don't need the has(bootstrapServerOpt) check.

@showuon showuon Apr 2, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You're right.

CommandLineUtils.checkInvalidArgsSet(parser, options, Set(bootstrapServerOpt, configOpt), Set(alterOpt),
Expand All @@ -767,9 +599,9 @@ object TopicCommand extends Logging {
CommandLineUtils.checkInvalidArgs(parser, options, reportUnderReplicatedPartitionsOpt,
allTopicLevelOpts -- Set(describeOpt) ++ allReplicationReportOpts - reportUnderReplicatedPartitionsOpt + topicsWithOverridesOpt)
CommandLineUtils.checkInvalidArgs(parser, options, reportUnderMinIsrPartitionsOpt,
allTopicLevelOpts -- Set(describeOpt) ++ allReplicationReportOpts - reportUnderMinIsrPartitionsOpt + topicsWithOverridesOpt + zkConnectOpt)
allTopicLevelOpts -- Set(describeOpt) ++ allReplicationReportOpts - reportUnderMinIsrPartitionsOpt + topicsWithOverridesOpt)
CommandLineUtils.checkInvalidArgs(parser, options, reportAtMinIsrPartitionsOpt,
allTopicLevelOpts -- Set(describeOpt) ++ allReplicationReportOpts - reportAtMinIsrPartitionsOpt + topicsWithOverridesOpt + zkConnectOpt)
allTopicLevelOpts -- Set(describeOpt) ++ allReplicationReportOpts - reportAtMinIsrPartitionsOpt + topicsWithOverridesOpt)
CommandLineUtils.checkInvalidArgs(parser, options, reportUnavailablePartitionsOpt,
allTopicLevelOpts -- Set(describeOpt) ++ allReplicationReportOpts - reportUnavailablePartitionsOpt + topicsWithOverridesOpt)
CommandLineUtils.checkInvalidArgs(parser, options, topicsWithOverridesOpt,
Expand Down
Loading