diff --git a/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala b/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala index 9937558abe32a..186a4313af08e 100755 --- a/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala +++ b/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala @@ -22,16 +22,14 @@ import java.util.concurrent.ExecutionException import kafka.common.AdminCommandFailedException import kafka.log.LogConfig -import kafka.server.{ConfigType, DynamicConfig} +import kafka.server.DynamicConfig import kafka.utils.{CommandDefaultOptions, CommandLineUtils, CoreUtils, Exit, Json, Logging} import kafka.utils.Implicits._ import kafka.utils.json.JsonValue -import kafka.zk.{AdminZkClient, KafkaZkClient} import org.apache.kafka.clients.admin.AlterConfigOp.OpType import org.apache.kafka.clients.admin.{Admin, AdminClientConfig, AlterConfigOp, ConfigEntry, NewPartitionReassignment, PartitionReassignment, TopicDescription} import org.apache.kafka.common.config.ConfigResource import org.apache.kafka.common.errors.{ReplicaNotAvailableException, UnknownTopicOrPartitionException} -import org.apache.kafka.common.security.JaasUtils import org.apache.kafka.common.utils.{Time, Utils} import org.apache.kafka.common.{KafkaException, KafkaFuture, TopicPartition, TopicPartitionReplica} @@ -194,33 +192,18 @@ object ReassignPartitionsCommand extends Logging { def main(args: Array[String]): Unit = { val opts = validateAndParseArgs(args) - var toClose: Option[AutoCloseable] = None var failed = true + var adminClient: Admin = null try { - if (opts.options.has(opts.bootstrapServerOpt)) { - if (opts.options.has(opts.zkConnectOpt)) { - println("Warning: ignoring deprecated --zookeeper option because " + - "--bootstrap-server was specified. The --zookeeper option will " + - "be removed in a future version of Kafka.") - } - val props = if (opts.options.has(opts.commandConfigOpt)) - Utils.loadProps(opts.options.valueOf(opts.commandConfigOpt)) - else - new util.Properties() - props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, opts.options.valueOf(opts.bootstrapServerOpt)) - props.putIfAbsent(AdminClientConfig.CLIENT_ID_CONFIG, "reassign-partitions-tool") - val adminClient = Admin.create(props) - toClose = Some(adminClient) - handleAction(adminClient, opts) - } else { - println("Warning: --zookeeper is deprecated, and will be removed in a future " + - "version of Kafka.") - val zkClient = KafkaZkClient(opts.options.valueOf(opts.zkConnectOpt), - JaasUtils.isZkSaslEnabled, 30000, 30000, Int.MaxValue, Time.SYSTEM) - toClose = Some(zkClient) - handleAction(zkClient, opts) - } + val props = if (opts.options.has(opts.commandConfigOpt)) + Utils.loadProps(opts.options.valueOf(opts.commandConfigOpt)) + else + new util.Properties() + props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, opts.options.valueOf(opts.bootstrapServerOpt)) + props.putIfAbsent(AdminClientConfig.CLIENT_ID_CONFIG, "reassign-partitions-tool") + adminClient = Admin.create(props) + handleAction(adminClient, opts) failed = false } catch { case e: TerseReassignmentFailureException => @@ -229,9 +212,10 @@ object ReassignPartitionsCommand extends Logging { println("Error: " + e.getMessage) println(Utils.stackTrace(e)) } finally { - // Close the AdminClient or ZooKeeper client, as appropriate. // It's good to do this after printing any error stack trace. - toClose.foreach(_.close()) + if (adminClient != null) { + adminClient.close() + } } // If the command failed, exit with a non-zero exit code. if (failed) { @@ -269,26 +253,6 @@ object ReassignPartitionsCommand extends Logging { } } - private def handleAction(zkClient: KafkaZkClient, - opts: ReassignPartitionsCommandOptions): Unit = { - if (opts.options.has(opts.verifyOpt)) { - verifyAssignment(zkClient, - Utils.readFileAsString(opts.options.valueOf(opts.reassignmentJsonFileOpt)), - opts.options.has(opts.preserveThrottlesOpt)) - } else if (opts.options.has(opts.generateOpt)) { - generateAssignment(zkClient, - Utils.readFileAsString(opts.options.valueOf(opts.topicsToMoveJsonFileOpt)), - opts.options.valueOf(opts.brokerListOpt), - !opts.options.has(opts.disableRackAware)) - } else if (opts.options.has(opts.executeOpt)) { - executeAssignment(zkClient, - Utils.readFileAsString(opts.options.valueOf(opts.reassignmentJsonFileOpt)), - opts.options.valueOf(opts.interBrokerThrottleOpt)) - } else { - throw new RuntimeException("Unsupported action.") - } - } - /** * A result returned from verifyAssignment. * @@ -345,50 +309,6 @@ object ReassignPartitionsCommand extends Logging { (partStates, partsOngoing) } - /** - * The deprecated entry point for the --verify command. - * - * @param zkClient The ZooKeeper client to use. - * @param jsonString The JSON string to use for the topics and partitions to verify. - * @param preserveThrottles True if we should avoid changing topic or broker throttles. - * - * @return A result that is useful for testing. Note that anything that - * would require AdminClient to see will be left out of this result. - */ - def verifyAssignment(zkClient: KafkaZkClient, jsonString: String, preserveThrottles: Boolean) - : VerifyAssignmentResult = { - val (targetParts, targetLogDirs) = parsePartitionReassignmentData(jsonString) - if (targetLogDirs.nonEmpty) { - throw new AdminCommandFailedException("bootstrap-server needs to be provided when " + - "replica reassignments are present.") - } - println("Warning: because you are using the deprecated --zookeeper option, the results " + - "may be incomplete. Use --bootstrap-server instead for more accurate results.") - val (partStates, partsOngoing) = verifyPartitionAssignments(zkClient, targetParts.toMap) - if (!partsOngoing && !preserveThrottles) { - clearAllThrottles(zkClient, targetParts) - } - VerifyAssignmentResult(partStates, partsOngoing, Map.empty, false) - } - - /** - * Verify the partition reassignments specified by the user. - * - * @param zkClient The ZooKeeper client to use. - * @param targets The partition reassignments specified by the user. - * - * @return A tuple of partition states and whether there are any - * ongoing reassignments found in the legacy reassign - * partitions ZNode. - */ - def verifyPartitionAssignments(zkClient: KafkaZkClient, - targets: Map[TopicPartition, Seq[Int]]) - : (Map[TopicPartition, PartitionReassignmentState], Boolean) = { - val (partStates, partsOngoing) = findPartitionReassignmentStates(zkClient, targets) - println(partitionReassignmentStatesToString(partStates)) - (partStates, partsOngoing) - } - def compareTopicPartitions(a: TopicPartition, b: TopicPartition): Boolean = { (a.topic(), a.partition()) < (b.topic(), b.partition()) } @@ -492,32 +412,6 @@ object ReassignPartitionsCommand extends Logging { } } - /** - * Find the state of the specified partition reassignments. - * - * @param zkClient The ZooKeeper client to use. - * @param targetReassignments The reassignments we want to learn about. - * - * @return A tuple containing the reassignment states for each topic - * partition, plus whether there are any ongoing reassignments - * found in the legacy reassign partitions znode. - */ - def findPartitionReassignmentStates(zkClient: KafkaZkClient, - targetReassignments: Map[TopicPartition, Seq[Int]]) - : (Map[TopicPartition, PartitionReassignmentState], Boolean) = { - val partitionsBeingReassigned = zkClient.getPartitionReassignment - val results = new mutable.HashMap[TopicPartition, PartitionReassignmentState]() - targetReassignments.groupBy(_._1.topic).forKeyValue { (topic, partitions) => - val replicasForTopic = zkClient.getReplicaAssignmentForTopics(Set(topic)) - partitions.forKeyValue { (partition, targetReplicas) => - val currentReplicas = replicasForTopic.getOrElse(partition, Seq()) - results.put(partition, new PartitionReassignmentState( - currentReplicas, targetReplicas, !partitionsBeingReassigned.contains(partition))) - } - } - (results, partitionsBeingReassigned.nonEmpty) - } - /** * Verify the replica reassignments specified by the user. * @@ -635,26 +529,6 @@ object ReassignPartitionsCommand extends Logging { clearTopicLevelThrottles(adminClient, topics) } - /** - * Clear all topic-level and broker-level throttles. - * - * @param zkClient The ZooKeeper client to use. - * @param targetParts The target partitions loaded from the JSON file. - */ - def clearAllThrottles(zkClient: KafkaZkClient, - targetParts: Seq[(TopicPartition, Seq[Int])]): Unit = { - val activeBrokers = zkClient.getAllBrokersInCluster.map(_.id).toSet - val brokers = activeBrokers ++ targetParts.flatMap(_._2).toSet - println("Clearing broker-level throttles on broker%s %s".format( - if (brokers.size == 1) "" else "s", brokers.mkString(","))) - clearBrokerLevelThrottles(zkClient, brokers) - - val topics = targetParts.map(_._1.topic()).toSet - println("Clearing topic-level throttles on topic%s %s".format( - if (topics.size == 1) "" else "s", topics.mkString(","))) - clearTopicLevelThrottles(zkClient, topics) - } - /** * Clear all throttles which have been set at the broker level. * @@ -672,22 +546,6 @@ object ReassignPartitionsCommand extends Logging { adminClient.incrementalAlterConfigs(configOps).all().get() } - /** - * Clear all throttles which have been set at the broker level. - * - * @param zkClient The ZooKeeper client to use. - * @param brokers The brokers to clear the throttles for. - */ - def clearBrokerLevelThrottles(zkClient: KafkaZkClient, brokers: Set[Int]): Unit = { - val adminZkClient = new AdminZkClient(zkClient) - for (brokerId <- brokers) { - val configs = adminZkClient.fetchEntityConfig(ConfigType.Broker, brokerId.toString) - if (brokerLevelThrottles.flatMap(throttle => Option(configs.remove(throttle))).nonEmpty) { - adminZkClient.changeBrokerConfig(Seq(brokerId), configs) - } - } - } - /** * Clear the reassignment throttles for the specified topics. * @@ -705,22 +563,6 @@ object ReassignPartitionsCommand extends Logging { adminClient.incrementalAlterConfigs(configOps).all().get() } - /** - * Clear the reassignment throttles for the specified topics. - * - * @param zkClient The ZooKeeper client to use. - * @param topics The topics to clear the throttles for. - */ - def clearTopicLevelThrottles(zkClient: KafkaZkClient, topics: Set[String]): Unit = { - val adminZkClient = new AdminZkClient(zkClient) - for (topic <- topics) { - val configs = adminZkClient.fetchEntityConfig(ConfigType.Topic, topic) - if (topicLevelThrottles.flatMap(throttle => Option(configs.remove(throttle))).nonEmpty) { - adminZkClient.changeTopicConfig(topic, configs) - } - } - } - /** * The entry point for the --generate command. * @@ -749,34 +591,6 @@ object ReassignPartitionsCommand extends Logging { (proposedAssignments, currentAssignments) } - /** - * The legacy entry point for the --generate command. - * - * @param zkClient The ZooKeeper client to use. - * @param reassignmentJson The JSON string to use for the topics to reassign. - * @param brokerListString The comma-separated string of broker IDs to use. - * @param enableRackAwareness True if rack-awareness should be enabled. - * - * @return A tuple containing the proposed assignment and the - * current assignment. - */ - def generateAssignment(zkClient: KafkaZkClient, - reassignmentJson: String, - brokerListString: String, - enableRackAwareness: Boolean) - : (Map[TopicPartition, Seq[Int]], Map[TopicPartition, Seq[Int]]) = { - val (brokersToReassign, topicsToReassign) = - parseGenerateAssignmentArgs(reassignmentJson, brokerListString) - val currentAssignments = zkClient.getReplicaAssignmentForTopics(topicsToReassign.toSet) - val brokerMetadatas = getBrokerMetadata(zkClient, brokersToReassign, enableRackAwareness) - val proposedAssignments = calculateAssignment(currentAssignments, brokerMetadatas) - println("Current partition replica assignment\n%s\n". - format(formatAsReassignmentJson(currentAssignments, Map.empty))) - println("Proposed partition reassignment configuration\n%s". - format(formatAsReassignmentJson(proposedAssignments, Map.empty))) - (proposedAssignments, currentAssignments) - } - /** * Calculate the new partition assignments to suggest in --generate. * @@ -888,25 +702,6 @@ object ReassignPartitionsCommand extends Logging { results } - /** - * Find the metadata for some brokers. - * - * @param zkClient The ZooKeeper client to use. - * @param brokers The brokers to gather metadata about. - * @param enableRackAwareness True if we should return rack information, and throw an - * exception if it is inconsistent. - * - * @return The metadata for each broker that was found. - * Brokers that were not found will be omitted. - */ - def getBrokerMetadata(zkClient: KafkaZkClient, - brokers: Seq[Int], - enableRackAwareness: Boolean): Seq[BrokerMetadata] = { - val adminZkClient = new AdminZkClient(zkClient) - adminZkClient.getBrokerMetadatas(if (enableRackAwareness) - RackAwareMode.Enforced else RackAwareMode.Disabled, Some(brokers)) - } - /** * Parse and validate data gathered from the command-line for --generate * In particular, we parse the JSON and validate that duplicate brokers and @@ -1088,56 +883,6 @@ object ReassignPartitionsCommand extends Logging { } } - /** - * The entry point for the --execute command. - * - * @param zkClient The ZooKeeper client to use. - * @param reassignmentJson The JSON string to use for the topics to reassign. - * @param interBrokerThrottle The inter-broker throttle to use, or a negative number - * to skip using a throttle. - */ - def executeAssignment(zkClient: KafkaZkClient, - reassignmentJson: String, - interBrokerThrottle: Long): Unit = { - val (proposedParts, proposedReplicas) = parseExecuteAssignmentArgs(reassignmentJson) - if (proposedReplicas.nonEmpty) { - throw new AdminCommandFailedException("bootstrap-server needs to be provided when " + - "replica reassignments are present.") - } - verifyReplicasAndBrokersInAssignment(zkClient, proposedParts) - - // Check for the presence of the legacy partition reassignment ZNode. This actually - // won't detect all rebalances... only ones initiated by the legacy method. - // This is a limitation of the legacy ZK API. - val reassignPartitionsInProgress = zkClient.reassignPartitionsInProgress - if (reassignPartitionsInProgress) { - // Note: older versions of this tool would modify the broker quotas here (but not - // topic quotas, for some reason). Since it might interfere with other ongoing - // reassignments, this behavior was dropped as part of the KIP-455 changes. The - // user can still alter existing throttles by resubmitting the current reassignment - // and providing the --additional flag. - throw new TerseReassignmentFailureException(cannotExecuteBecauseOfExistingMessage) - } - val currentParts = zkClient.getReplicaAssignmentForTopics( - proposedParts.map(_._1.topic()).toSet) - println(currentPartitionReplicaAssignmentToString(proposedParts, currentParts)) - - if (interBrokerThrottle >= 0) { - println(youMustRunVerifyPeriodicallyMessage) - val moveMap = calculateProposedMoveMap(Map.empty, proposedParts, currentParts) - val leaderThrottles = calculateLeaderThrottles(moveMap) - val followerThrottles = calculateFollowerThrottles(moveMap) - modifyTopicThrottles(zkClient, leaderThrottles, followerThrottles) - val reassigningBrokers = calculateReassigningBrokers(moveMap) - modifyBrokerThrottles(zkClient, reassigningBrokers, interBrokerThrottle) - println(s"The inter-broker throttle limit was set to ${interBrokerThrottle} B/s") - } - zkClient.createPartitionReassignment(proposedParts) - println("Successfully started partition reassignment%s for %s".format( - if (proposedParts.size == 1) "" else "s", - proposedParts.keySet.toBuffer.sortWith(compareTopicPartitions).mkString(","))) - } - /** * Return the string which we want to print to describe the current partition assignment. * @@ -1154,31 +899,6 @@ object ReassignPartitionsCommand extends Logging { "--reassignment-json-file option during rollback") } - /** - * Verify that the replicas and brokers referenced in the given partition assignment actually - * exist. This is necessary when using the deprecated ZK API, since ZooKeeper itself can't - * validate what we're applying. - * - * @param zkClient The ZooKeeper client to use. - * @param proposedParts The partition assignment. - */ - def verifyReplicasAndBrokersInAssignment(zkClient: KafkaZkClient, - proposedParts: Map[TopicPartition, Seq[Int]]): Unit = { - // check that all partitions in the proposed assignment exist in the cluster - val proposedTopics = proposedParts.map { case (tp, _) => tp.topic } - val existingAssignment = zkClient.getReplicaAssignmentForTopics(proposedTopics.toSet) - val nonExistentPartitions = proposedParts.map { case (tp, _) => tp }.filterNot(existingAssignment.contains) - if (nonExistentPartitions.nonEmpty) - throw new AdminCommandFailedException("The proposed assignment contains non-existent partitions: " + - nonExistentPartitions) - - // check that all brokers in the proposed assignment exist in the cluster - val existingBrokerIDs = zkClient.getSortedBrokerList - val nonExistingBrokerIDs = proposedParts.toMap.values.flatten.filterNot(existingBrokerIDs.contains).toSet - if (nonExistingBrokerIDs.nonEmpty) - throw new AdminCommandFailedException("The proposed assignment contains non-existent brokerIDs: " + nonExistingBrokerIDs.mkString(",")) - } - /** * Execute the given partition reassignments. * @@ -1375,26 +1095,6 @@ object ReassignPartitionsCommand extends Logging { adminClient.incrementalAlterConfigs(configs).all().get() } - /** - * Modify the topic configurations that control inter-broker throttling. - * - * @param zkClient The ZooKeeper client to use. - * @param leaderThrottles A map from topic names to leader throttle configurations. - * @param followerThrottles A map from topic names to follower throttle configurations. - */ - def modifyTopicThrottles(zkClient: KafkaZkClient, - leaderThrottles: Map[String, String], - followerThrottles: Map[String, String]): Unit = { - val adminZkClient = new AdminZkClient(zkClient) - val topicNames = leaderThrottles.keySet ++ followerThrottles.keySet - topicNames.foreach { topicName => - val configs = adminZkClient.fetchEntityConfig(ConfigType.Topic, topicName) - leaderThrottles.get(topicName).map(configs.put(topicLevelLeaderThrottle, _)) - followerThrottles.get(topicName).map(configs.put(topicLevelFollowerThrottle, _)) - adminZkClient.changeTopicConfig(topicName, configs) - } - } - private def modifyReassignmentThrottle(admin: Admin, moveMap: MoveMap, interBrokerThrottle: Long): Unit = { val leaderThrottles = calculateLeaderThrottles(moveMap) val followerThrottles = calculateFollowerThrottles(moveMap) @@ -1451,25 +1151,6 @@ object ReassignPartitionsCommand extends Logging { } } - /** - * Modify the broker-level configurations for leader and follower throttling. - * - * @param zkClient The ZooKeeper client to use. - * @param reassigningBrokers The brokers to reconfigure. - * @param interBrokerThrottle The throttle value to set. - */ - def modifyBrokerThrottles(zkClient: KafkaZkClient, - reassigningBrokers: Set[Int], - interBrokerThrottle: Long): Unit = { - val adminZkClient = new AdminZkClient(zkClient) - for (id <- reassigningBrokers) { - val configs = adminZkClient.fetchEntityConfig(ConfigType.Broker, id.toString) - configs.put(brokerLevelLeaderThrottle, interBrokerThrottle.toString) - configs.put(brokerLevelFollowerThrottle, interBrokerThrottle.toString) - adminZkClient.changeBrokerConfig(Seq(id), configs) - } - } - /** * Parse the reassignment JSON string passed to the --execute command. * @@ -1662,12 +1343,7 @@ object ReassignPartitionsCommand extends Logging { } val action = allActions(0) - // Check that we have either the --zookeeper option or the --bootstrap-server set. - // It would be nice to enforce that we can only have one of these options set at once. Unfortunately, - // previous versions of this tool supported setting both options together. To avoid breaking backwards - // compatibility, we will follow suit, for now. This issue will eventually be resolved when we remove - // the --zookeeper option. - if (!opts.options.has(opts.zkConnectOpt) && !opts.options.has(opts.bootstrapServerOpt)) + if (!opts.options.has(opts.bootstrapServerOpt)) CommandLineUtils.printUsageAndDie(opts.parser, "Please specify --bootstrap-server") // Make sure that we have all the required arguments for our action. @@ -1695,14 +1371,12 @@ object ReassignPartitionsCommand extends Logging { opts.bootstrapServerOpt, opts.commandConfigOpt, opts.preserveThrottlesOpt, - opts.zkConnectOpt ), opts.generateOpt -> Seq( opts.bootstrapServerOpt, opts.brokerListOpt, opts.commandConfigOpt, opts.disableRackAware, - opts.zkConnectOpt ), opts.executeOpt -> Seq( opts.additionalOpt, @@ -1711,7 +1385,6 @@ object ReassignPartitionsCommand extends Logging { opts.interBrokerThrottleOpt, opts.replicaAlterLogDirsThrottleOpt, opts.timeoutOpt, - opts.zkConnectOpt ), opts.cancelOpt -> Seq( opts.bootstrapServerOpt, @@ -1726,28 +1399,13 @@ object ReassignPartitionsCommand extends Logging { ) opts.options.specs.forEach(opt => { if (!opt.equals(action) && - !requiredArgs(action).contains(opt) && - !permittedArgs(action).contains(opt)) { + !requiredArgs(action).contains(opt) && + !permittedArgs(action).contains(opt)) { CommandLineUtils.printUsageAndDie(opts.parser, """Option "%s" can't be used with action "%s"""".format(opt, action)) } }) - if (!opts.options.has(opts.bootstrapServerOpt)) { - val bootstrapServerOnlyArgs = Seq( - opts.additionalOpt, - opts.cancelOpt, - opts.commandConfigOpt, - opts.replicaAlterLogDirsThrottleOpt, - opts.listOpt, - opts.timeoutOpt - ) - bootstrapServerOnlyArgs.foreach { - opt => if (opts.options.has(opt)) { - throw new RuntimeException("You must specify --bootstrap-server " + - """when using "%s"""".format(opt)) - } - } - } + opts } @@ -1775,7 +1433,8 @@ object ReassignPartitionsCommand extends Logging { sealed class ReassignPartitionsCommandOptions(args: Array[String]) extends CommandDefaultOptions(args) { // Actions - val verifyOpt = parser.accepts("verify", "Verify if the reassignment completed as specified by the --reassignment-json-file option. If there is a throttle engaged for the replicas specified, and the rebalance has completed, the throttle will be removed") + val verifyOpt = parser.accepts("verify", "Verify if the reassignment completed as specified by the " + + "--reassignment-json-file option. If there is a throttle engaged for the replicas specified, and the rebalance has completed, the throttle will be removed") val generateOpt = parser.accepts("generate", "Generate a candidate partition reassignment configuration." + " Note that this only generates a candidate assignment, it does not execute it.") val executeOpt = parser.accepts("execute", "Kick off the reassignment as specified by the --reassignment-json-file option.") @@ -1783,21 +1442,16 @@ object ReassignPartitionsCommand extends Logging { val listOpt = parser.accepts("list", "List all active partition reassignments.") // Arguments - val bootstrapServerOpt = parser.accepts("bootstrap-server", "the server(s) to use for bootstrapping. REQUIRED if " + - "an absolute path of the log directory is specified for any replica in the reassignment json file, " + - "or if --zookeeper is not given.") + val bootstrapServerOpt = parser.accepts("bootstrap-server", "REQUIRED: the server(s) to use for bootstrapping.") .withRequiredArg .describedAs("Server(s) to use for bootstrapping") .ofType(classOf[String]) + val commandConfigOpt = parser.accepts("command-config", "Property file containing configs to be passed to Admin Client.") .withRequiredArg .describedAs("Admin client property file") .ofType(classOf[String]) - val zkConnectOpt = parser.accepts("zookeeper", "DEPRECATED: The connection string for the zookeeper connection in the " + - "form host:port. Multiple URLS can be given to allow fail-over. Please use --bootstrap-server instead.") - .withRequiredArg - .describedAs("urls") - .ofType(classOf[String]) + val reassignmentJsonFileOpt = parser.accepts("reassignment-json-file", "The JSON file with the partition reassignment configuration" + "The format to use is - \n" + "{\"partitions\":\n\t[{\"topic\": \"foo\",\n\t \"partition\": 1,\n\t \"replicas\": [1,2,3],\n\t \"log_dirs\": [\"dir1\",\"dir2\",\"dir3\"] }],\n\"version\":1\n}\n" + diff --git a/core/src/test/scala/integration/kafka/admin/ReassignPartitionsIntegrationTest.scala b/core/src/test/scala/integration/kafka/admin/ReassignPartitionsIntegrationTest.scala index 07c5c2ad0bd1d..490d177a94119 100644 --- a/core/src/test/scala/integration/kafka/admin/ReassignPartitionsIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/admin/ReassignPartitionsIntegrationTest.scala @@ -24,7 +24,7 @@ import kafka.api.KAFKA_2_7_IV1 import kafka.server.{IsrChangePropagationConfig, KafkaConfig, KafkaServer, ZkIsrManager} import kafka.utils.Implicits._ import kafka.utils.TestUtils -import kafka.zk.{KafkaZkClient, ZooKeeperTestHarness} +import kafka.zk.ZooKeeperTestHarness import org.apache.kafka.clients.admin.{Admin, AdminClientConfig, AlterConfigOp, ConfigEntry, DescribeLogDirsResult, NewTopic} import org.apache.kafka.clients.producer.ProducerRecord import org.apache.kafka.common.config.ConfigResource @@ -115,8 +115,6 @@ class ReassignPartitionsIntegrationTest extends ZooKeeperTestHarness { ) waitForVerifyAssignment(cluster.adminClient, assignment, false, VerifyAssignmentResult(initialAssignment)) - waitForVerifyAssignment(zkClient, assignment, false, - VerifyAssignmentResult(initialAssignment)) // Execute the assignment runExecuteAssignment(cluster.adminClient, false, assignment, -1L, -1L) @@ -128,43 +126,18 @@ class ReassignPartitionsIntegrationTest extends ZooKeeperTestHarness { PartitionReassignmentState(Seq(3, 2, 0), Seq(3, 2, 0), true) ) - // When using --zookeeper, we aren't able to see the new-style assignment - assertFalse(runVerifyAssignment(zkClient, assignment, false).movesOngoing) + val verifyAssignmentResult = runVerifyAssignment(cluster.adminClient, assignment, false) + assertTrue(verifyAssignmentResult.partsOngoing) + assertFalse(verifyAssignmentResult.movesOngoing) // Wait for the assignment to complete - waitForVerifyAssignment(zkClient, assignment, false, + waitForVerifyAssignment(cluster.adminClient, assignment, false, VerifyAssignmentResult(finalAssignment)) assertEquals(unthrottledBrokerConfigs, describeBrokerLevelThrottles(unthrottledBrokerConfigs.keySet.toSeq)) } - /** - * Test running a quick reassignment with the --zookeeper option. - */ - @Test - def testLegacyReassignment(): Unit = { - cluster = new ReassignPartitionsTestCluster(zkConnect) - cluster.setup() - val assignment = """{"version":1,"partitions":""" + - """[{"topic":"foo","partition":0,"replicas":[0,1,3],"log_dirs":["any","any","any"]},""" + - """{"topic":"bar","partition":0,"replicas":[3,2,0],"log_dirs":["any","any","any"]}""" + - """]}""" - // Execute the assignment - runExecuteAssignment(zkClient, assignment, -1L) - val finalAssignment = Map( - new TopicPartition("foo", 0) -> - PartitionReassignmentState(Seq(0, 1, 3), Seq(0, 1, 3), true), - new TopicPartition("bar", 0) -> - PartitionReassignmentState(Seq(3, 2, 0), Seq(3, 2, 0), true) - ) - // Wait for the assignment to complete - waitForVerifyAssignment(cluster.adminClient, assignment, false, - VerifyAssignmentResult(finalAssignment)) - waitForVerifyAssignment(zkClient, assignment, false, - VerifyAssignmentResult(finalAssignment)) - } - @Test def testHighWaterMarkAfterPartitionReassignment(): Unit = { cluster = new ReassignPartitionsTestCluster(zkConnect) @@ -180,7 +153,7 @@ class ReassignPartitionsIntegrationTest extends ZooKeeperTestHarness { // Execute the assignment runExecuteAssignment(cluster.adminClient, false, assignment, -1L, -1L) val finalAssignment = Map(part -> - PartitionReassignmentState(Seq(3, 1, 2), Seq(3, 1, 2), true)) + PartitionReassignmentState(Seq(3, 1, 2), Seq(3, 1, 2), true)) // Wait for the assignment to complete waitForVerifyAssignment(cluster.adminClient, assignment, false, @@ -189,7 +162,7 @@ class ReassignPartitionsIntegrationTest extends ZooKeeperTestHarness { TestUtils.waitUntilTrue(() => { cluster.servers(3).replicaManager.onlinePartition(part). flatMap(_.leaderLogIfLocal).isDefined - }, "broker 3 should be the new leader", pause = 10L) + }, "broker 3 should be the new leader", pause = 10L) assertEquals(123L, cluster.servers(3).replicaManager.localLogOrException(part).highWatermark, s"Expected broker 3 to have the correct high water mark for the partition.") } @@ -248,7 +221,6 @@ class ReassignPartitionsIntegrationTest extends ZooKeeperTestHarness { new TopicPartition("baz", 2) -> PartitionReassignmentState(Seq(0, 2, 1), Seq(3, 2, 1), true)) assertEquals(VerifyAssignmentResult(initialAssignment), runVerifyAssignment(cluster.adminClient, assignment, false)) - assertEquals(VerifyAssignmentResult(initialAssignment), runVerifyAssignment(zkClient, assignment, false)) assertEquals(unthrottledBrokerConfigs, describeBrokerLevelThrottles(unthrottledBrokerConfigs.keySet.toSeq)) // Execute the assignment @@ -280,8 +252,6 @@ class ReassignPartitionsIntegrationTest extends ZooKeeperTestHarness { }, "Expected reassignment to complete.") waitForVerifyAssignment(cluster.adminClient, assignment, true, VerifyAssignmentResult(finalAssignment)) - waitForVerifyAssignment(zkClient, assignment, true, - VerifyAssignmentResult(finalAssignment)) // The throttles should still have been preserved, since we ran with --preserve-throttles waitForInterBrokerThrottle(Set(0, 1, 2, 3), interBrokerThrottle) // Now remove the throttles. @@ -340,12 +310,12 @@ class ReassignPartitionsIntegrationTest extends ZooKeeperTestHarness { VerifyAssignmentResult(Map( new TopicPartition("foo", 0) -> PartitionReassignmentState(Seq(0, 1, 3, 2), Seq(0, 1, 3), false), new TopicPartition("baz", 1) -> PartitionReassignmentState(Seq(0, 2, 3, 1), Seq(0, 2, 3), false)), - true, Map(), false)) + true, Map(), false)) // Cancel the reassignment. assertEquals((Set( - new TopicPartition("foo", 0), - new TopicPartition("baz", 1) - ), Set()), runCancelAssignment(cluster.adminClient, assignment, true)) + new TopicPartition("foo", 0), + new TopicPartition("baz", 1) + ), Set()), runCancelAssignment(cluster.adminClient, assignment, true)) // Broker throttles are still active because we passed --preserve-throttles waitForInterBrokerThrottle(Set(0, 1, 2, 3), interBrokerThrottle) // Cancelling the reassignment again should reveal nothing to cancel. @@ -433,32 +403,32 @@ class ReassignPartitionsIntegrationTest extends ZooKeeperTestHarness { // Check the output of --verify waitForVerifyAssignment(cluster.adminClient, reassignment.json, true, VerifyAssignmentResult(Map( - topicPartition -> PartitionReassignmentState(Seq(0, 1, 2), Seq(0, 1, 2), true) - ), false, Map( - new TopicPartitionReplica(topicPartition.topic, topicPartition.partition, 0) -> - ActiveMoveState(reassignment.currentDir, reassignment.targetDir, reassignment.targetDir) - ), true)) + topicPartition -> PartitionReassignmentState(Seq(0, 1, 2), Seq(0, 1, 2), true) + ), false, Map( + new TopicPartitionReplica(topicPartition.topic, topicPartition.partition, 0) -> + ActiveMoveState(reassignment.currentDir, reassignment.targetDir, reassignment.targetDir) + ), true)) waitForLogDirThrottle(Set(0), logDirThrottle) // Remove the throttle cluster.adminClient.incrementalAlterConfigs(Collections.singletonMap( new ConfigResource(ConfigResource.Type.BROKER, "0"), Collections.singletonList(new AlterConfigOp( - new ConfigEntry(brokerLevelLogDirThrottle, ""), AlterConfigOp.OpType.DELETE)))). - all().get() + new ConfigEntry(brokerLevelLogDirThrottle, ""), AlterConfigOp.OpType.DELETE)))) + .all().get() waitForBrokerLevelThrottles(unthrottledBrokerConfigs) // Wait for the directory movement to complete. waitForVerifyAssignment(cluster.adminClient, reassignment.json, true, - VerifyAssignmentResult(Map( - topicPartition -> PartitionReassignmentState(Seq(0, 1, 2), Seq(0, 1, 2), true) - ), false, Map( - new TopicPartitionReplica(topicPartition.topic, topicPartition.partition, 0) -> - CompletedMoveState(reassignment.targetDir) - ), false)) + VerifyAssignmentResult(Map( + topicPartition -> PartitionReassignmentState(Seq(0, 1, 2), Seq(0, 1, 2), true) + ), false, Map( + new TopicPartitionReplica(topicPartition.topic, topicPartition.partition, 0) -> + CompletedMoveState(reassignment.targetDir) + ), false)) val info1 = new BrokerDirs(cluster.adminClient.describeLogDirs(0.to(4). - map(_.asInstanceOf[Integer]).asJavaCollection), 0) + map(_.asInstanceOf[Integer]).asJavaCollection), 0) assertEquals(reassignment.targetDir, info1.curLogDirs.getOrElse(topicPartition, "")) } @@ -540,7 +510,8 @@ class ReassignPartitionsIntegrationTest extends ZooKeeperTestHarness { verifyAssignment(adminClient, jsonString, preserveThrottles) } - private def waitForVerifyAssignment(adminClient: Admin, jsonString: String, + private def waitForVerifyAssignment(adminClient: Admin, + jsonString: String, preserveThrottles: Boolean, expectedResult: VerifyAssignmentResult): Unit = { var latestResult: VerifyAssignmentResult = null @@ -552,26 +523,6 @@ class ReassignPartitionsIntegrationTest extends ZooKeeperTestHarness { s"The latest result was ${latestResult}", pause = 10L) } - private def runVerifyAssignment(zkClient: KafkaZkClient, jsonString: String, - preserveThrottles: Boolean) = { - println(s"==> verifyAssignment(zkClient, jsonString=${jsonString})") - verifyAssignment(zkClient, jsonString, preserveThrottles) - } - - private def waitForVerifyAssignment(zkClient: KafkaZkClient, jsonString: String, - preserveThrottles: Boolean, - expectedResult: VerifyAssignmentResult): Unit = { - var latestResult: VerifyAssignmentResult = null - TestUtils.waitUntilTrue( - () => { - println(s"==> verifyAssignment(zkClient, jsonString=${jsonString}, " + - s"preserveThrottles=${preserveThrottles})") - latestResult = verifyAssignment(zkClient, jsonString, preserveThrottles) - expectedResult.equals(latestResult) - }, s"Timed out waiting for verifyAssignment result ${expectedResult}. " + - s"The latest result was ${latestResult}", pause = 10L) - } - private def runExecuteAssignment(adminClient: Admin, additional: Boolean, reassignmentJson: String, @@ -585,15 +536,6 @@ class ReassignPartitionsIntegrationTest extends ZooKeeperTestHarness { interBrokerThrottle, replicaAlterLogDirsThrottle) } - private def runExecuteAssignment(zkClient: KafkaZkClient, - reassignmentJson: String, - interBrokerThrottle: Long) = { - println(s"==> executeAssignment(adminClient, " + - s"reassignmentJson=${reassignmentJson}, " + - s"interBrokerThrottle=${interBrokerThrottle})") - executeAssignment(zkClient, reassignmentJson, interBrokerThrottle) - } - private def runCancelAssignment(adminClient: Admin, jsonString: String, preserveThrottles: Boolean) = { println(s"==> cancelAssignment(adminClient, jsonString=${jsonString})") diff --git a/core/src/test/scala/unit/kafka/admin/ReassignPartitionsCommandArgsTest.scala b/core/src/test/scala/unit/kafka/admin/ReassignPartitionsCommandArgsTest.scala index d0c13ac143759..13fc262ec4e72 100644 --- a/core/src/test/scala/unit/kafka/admin/ReassignPartitionsCommandArgsTest.scala +++ b/core/src/test/scala/unit/kafka/admin/ReassignPartitionsCommandArgsTest.scala @@ -23,6 +23,8 @@ import org.junit.jupiter.api.{AfterEach, BeforeEach, Test, Timeout} @Timeout(60) class ReassignPartitionsCommandArgsTest { + val missingBootstrapServerMsg = "Please specify --bootstrap-server" + @BeforeEach def setUp(): Unit = { Exit.setExitProcedure((_, message) => throw new IllegalArgumentException(message.orNull)) @@ -53,15 +55,6 @@ class ReassignPartitionsCommandArgsTest { ReassignPartitionsCommand.validateAndParseArgs(args) } - @Test - def shouldCorrectlyParseValidMinimumLegacyExecuteOptions(): Unit = { - val args = Array( - "--zookeeper", "localhost:1234", - "--execute", - "--reassignment-json-file", "myfile.json") - ReassignPartitionsCommand.validateAndParseArgs(args) - } - @Test def shouldCorrectlyParseValidMinimumVerifyOptions(): Unit = { val args = Array( @@ -71,15 +64,6 @@ class ReassignPartitionsCommandArgsTest { ReassignPartitionsCommand.validateAndParseArgs(args) } - @Test - def shouldCorrectlyParseValidMinimumLegacyVerifyOptions(): Unit = { - val args = Array( - "--zookeeper", "localhost:1234", - "--verify", - "--reassignment-json-file", "myfile.json") - ReassignPartitionsCommand.validateAndParseArgs(args) - } - @Test def shouldAllowThrottleOptionOnExecute(): Unit = { val args = Array( @@ -177,7 +161,7 @@ class ReassignPartitionsCommandArgsTest { def testMissingBootstrapServerArgumentForExecute(): Unit = { val args = Array( "--execute") - shouldFailWith("Please specify --bootstrap-server", args) + shouldFailWith(missingBootstrapServerMsg, args) } ///// Test --generate @@ -230,15 +214,10 @@ class ReassignPartitionsCommandArgsTest { } @Test - def testInvalidCommandConfigArgumentForLegacyGenerate(): Unit = { - val args = Array( - "--zookeeper", "localhost:1234", - "--generate", - "--broker-list", "101,102", - "--topics-to-move-json-file", "myfile.json", - "--command-config", "/tmp/command-config.properties" - ) - shouldFailWith("You must specify --bootstrap-server when using \"[command-config]\"", args) + def shouldPrintHelpTextIfHelpArg(): Unit = { + val args: Array[String]= Array("--help") + // note, this is not actually a failed case, it's just we share the same `printUsageAndDie` method when wrong arg received + shouldFailWith(ReassignPartitionsCommand.helpText, args) } ///// Test --verify @@ -291,7 +270,7 @@ class ReassignPartitionsCommandArgsTest { def shouldNotAllowCancelWithoutBootstrapServerOption(): Unit = { val args = Array( "--cancel") - shouldFailWith("Please specify --bootstrap-server", args) + shouldFailWith(missingBootstrapServerMsg, args) } @Test @@ -302,13 +281,4 @@ class ReassignPartitionsCommandArgsTest { "--preserve-throttles") shouldFailWith("Missing required argument \"[reassignment-json-file]\"", args) } - - ///// Test --list - @Test - def shouldNotAllowZooKeeperWithListOption(): Unit = { - val args = Array( - "--list", - "--zookeeper", "localhost:1234") - shouldFailWith("Option \"[zookeeper]\" can't be used with action \"[list]\"", args) - } } diff --git a/docs/upgrade.html b/docs/upgrade.html index bfb38abb603b6..70ca0258db550 100644 --- a/docs/upgrade.html +++ b/docs/upgrade.html @@ -43,7 +43,8 @@
Notable changes in 3 AclBindingFilter.
  • The Admin.electedPreferredLeaders() methods were removed. Please use Admin.electLeaders instead.
  • The kafka-preferred-replica-election command line tool was removed. Please use kafka-leader-election instead.
    • -
  • The --zookeeper option was removed from the bin/kafka-topics.sh command line tool. Please use --bootstrap-server instead.
    • +
  • The --zookeeper option was removed from the bin/kafka-topics.sh and bin/kafka-reassign-partitions.sh command line tools. + Please use --bootstrap-server instead.
  • The ConfigEntry constructor was removed (KAFKA-12577). Please use the remaining public constructor instead.
  • The config value default for the client config client.dns.lookup has been removed. In the unlikely