abbccdda commented on a change in pull request #9409: URL: https://github.com/apache/kafka/pull/9409#discussion_r503443796
########## File path: core/src/main/scala/kafka/admin/FeatureCommand.scala ########## @@ -0,0 +1,350 @@ +/** + * 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.server.BrokerFeatures +import kafka.utils.{CommandDefaultOptions, CommandLineUtils, Exit} +import org.apache.kafka.clients.CommonClientConfigs +import org.apache.kafka.clients.admin.{Admin, DescribeFeaturesOptions, FeatureMetadata, FeatureUpdate, UpdateFeaturesOptions} +import org.apache.kafka.common.feature.{Features, SupportedVersionRange} +import org.apache.kafka.common.utils.Utils + +import java.util.Properties +import scala.collection.Seq +import scala.collection.immutable.ListMap +import scala.jdk.CollectionConverters._ + +import joptsimple.OptionSpec + +object FeatureCommand { + + def main(args: Array[String]): Unit = { + val opts = new FeatureCommandOptions(args) + val featureApis = new FeatureApis(opts) + var exitCode = 0 + try { + featureApis.execute() + } catch { + case e: IllegalArgumentException => + printException(e) + opts.parser.printHelpOn(System.err) + exitCode = 1 + case e: Throwable => + printException(e) + exitCode = 1 + } finally { + featureApis.close() + Exit.exit(exitCode) + } + } + + private def printException(exception: Throwable): Unit = { + System.err.println("\nError encountered when executing command: " + Utils.stackTrace(exception)) Review comment: Is the `\n` necessary? ########## File path: core/src/main/scala/kafka/admin/FeatureCommand.scala ########## @@ -0,0 +1,350 @@ +/** + * 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.server.BrokerFeatures +import kafka.utils.{CommandDefaultOptions, CommandLineUtils, Exit} +import org.apache.kafka.clients.CommonClientConfigs +import org.apache.kafka.clients.admin.{Admin, DescribeFeaturesOptions, FeatureMetadata, FeatureUpdate, UpdateFeaturesOptions} +import org.apache.kafka.common.feature.{Features, SupportedVersionRange} +import org.apache.kafka.common.utils.Utils + +import java.util.Properties +import scala.collection.Seq +import scala.collection.immutable.ListMap +import scala.jdk.CollectionConverters._ + +import joptsimple.OptionSpec + +object FeatureCommand { + + def main(args: Array[String]): Unit = { + val opts = new FeatureCommandOptions(args) + val featureApis = new FeatureApis(opts) + var exitCode = 0 + try { + featureApis.execute() + } catch { + case e: IllegalArgumentException => + printException(e) + opts.parser.printHelpOn(System.err) + exitCode = 1 + case e: Throwable => + printException(e) + exitCode = 1 + } finally { + featureApis.close() + Exit.exit(exitCode) + } + } + + private def printException(exception: Throwable): Unit = { + System.err.println("\nError encountered when executing command: " + Utils.stackTrace(exception)) + } +} + +class UpdateFeaturesException(message: String) extends RuntimeException(message) + +/** + * A class that provides necessary APIs to bridge the Admin client feature APIs with the CLI tool. + * + * @param opts the CLI options + */ +class FeatureApis(var opts: FeatureCommandOptions) { + private var supportedFeatures = BrokerFeatures.createDefault().supportedFeatures + private val adminClient = createAdminClient() + + private def pad(op: String): String = { + f"$op%11s" + } + + private val addOp = pad("[Add]") + private val upgradeOp = pad("[Upgrade]") + private val deleteOp = pad("[Delete]") + private val downgradeOp = pad("[Downgrade]") + + // For testing only. + def setSupportedFeatures(newFeatures: Features[SupportedVersionRange]): Unit = { + supportedFeatures = newFeatures + } + + // For testing only. + def setOptions(newOpts: FeatureCommandOptions): Unit = { + opts = newOpts + } + + private def describeFeatures(sendRequestToController: Boolean): FeatureMetadata = { + val options = new DescribeFeaturesOptions().sendRequestToController(sendRequestToController) + adminClient.describeFeatures(options).featureMetadata().get() + } + + /** + * Describes the supported and finalized features. If the --from-controller CLI option + * is provided, then the request is issued only to the controller, otherwise the request is issued + * to any of the provided bootstrap servers. + */ + def describeFeatures(): Unit = { + val result = describeFeatures(opts.hasFromControllerOption) + val features = result.supportedFeatures.asScala.keys.toSet ++ result.finalizedFeatures.asScala.keys.toSet + features.toList.sorted.foreach { + case feature => + val (finalizedMinVersionLevel, finalizedMaxVersionLevel) = { + val finalizedVersionRange = result.finalizedFeatures.get(feature) + if (finalizedVersionRange == null) { + ("-", "-") + } else { + (finalizedVersionRange.minVersionLevel, finalizedVersionRange.maxVersionLevel) + } + } + val epoch = { + if (result.finalizedFeaturesEpoch.isPresent) { + result.finalizedFeaturesEpoch.get.toString + } else { + "-" + } + } + val (supportedMinVersion, supportedMaxVersion) = { + val supportedVersionRange = result.supportedFeatures.get(feature) + if (supportedVersionRange == null) { + ("-", "-") + } else { + (supportedVersionRange.minVersion, supportedVersionRange.maxVersion) + } + } + print(s"Feature: $feature") + print(s"\tSupportedMinVersion: $supportedMinVersion") + print(s"\tSupportedMaxVersion: $supportedMaxVersion") + print(s"\tFinalizedMinVersionLevel: $finalizedMinVersionLevel") Review comment: nit: we could do the comment immediately after the version get extracted, just like moving L132 to be after L113 ########## File path: core/src/main/scala/kafka/admin/FeatureCommand.scala ########## @@ -0,0 +1,350 @@ +/** + * 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.server.BrokerFeatures +import kafka.utils.{CommandDefaultOptions, CommandLineUtils, Exit} +import org.apache.kafka.clients.CommonClientConfigs +import org.apache.kafka.clients.admin.{Admin, DescribeFeaturesOptions, FeatureMetadata, FeatureUpdate, UpdateFeaturesOptions} +import org.apache.kafka.common.feature.{Features, SupportedVersionRange} +import org.apache.kafka.common.utils.Utils + +import java.util.Properties +import scala.collection.Seq +import scala.collection.immutable.ListMap +import scala.jdk.CollectionConverters._ + +import joptsimple.OptionSpec + +object FeatureCommand { + + def main(args: Array[String]): Unit = { + val opts = new FeatureCommandOptions(args) + val featureApis = new FeatureApis(opts) + var exitCode = 0 + try { + featureApis.execute() + } catch { + case e: IllegalArgumentException => + printException(e) + opts.parser.printHelpOn(System.err) + exitCode = 1 + case e: Throwable => + printException(e) + exitCode = 1 + } finally { + featureApis.close() + Exit.exit(exitCode) + } + } + + private def printException(exception: Throwable): Unit = { + System.err.println("\nError encountered when executing command: " + Utils.stackTrace(exception)) + } +} + +class UpdateFeaturesException(message: String) extends RuntimeException(message) + +/** + * A class that provides necessary APIs to bridge the Admin client feature APIs with the CLI tool. + * + * @param opts the CLI options + */ +class FeatureApis(var opts: FeatureCommandOptions) { + private var supportedFeatures = BrokerFeatures.createDefault().supportedFeatures + private val adminClient = createAdminClient() + + private def pad(op: String): String = { + f"$op%11s" + } + + private val addOp = pad("[Add]") + private val upgradeOp = pad("[Upgrade]") + private val deleteOp = pad("[Delete]") + private val downgradeOp = pad("[Downgrade]") + + // For testing only. + def setSupportedFeatures(newFeatures: Features[SupportedVersionRange]): Unit = { + supportedFeatures = newFeatures + } + + // For testing only. + def setOptions(newOpts: FeatureCommandOptions): Unit = { + opts = newOpts + } + + private def describeFeatures(sendRequestToController: Boolean): FeatureMetadata = { + val options = new DescribeFeaturesOptions().sendRequestToController(sendRequestToController) + adminClient.describeFeatures(options).featureMetadata().get() + } + + /** + * Describes the supported and finalized features. If the --from-controller CLI option + * is provided, then the request is issued only to the controller, otherwise the request is issued + * to any of the provided bootstrap servers. + */ + def describeFeatures(): Unit = { + val result = describeFeatures(opts.hasFromControllerOption) + val features = result.supportedFeatures.asScala.keys.toSet ++ result.finalizedFeatures.asScala.keys.toSet + features.toList.sorted.foreach { + case feature => Review comment: nit: case not necessary ########## File path: core/src/main/scala/kafka/admin/FeatureCommand.scala ########## @@ -0,0 +1,350 @@ +/** + * 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.server.BrokerFeatures +import kafka.utils.{CommandDefaultOptions, CommandLineUtils, Exit} +import org.apache.kafka.clients.CommonClientConfigs +import org.apache.kafka.clients.admin.{Admin, DescribeFeaturesOptions, FeatureMetadata, FeatureUpdate, UpdateFeaturesOptions} +import org.apache.kafka.common.feature.{Features, SupportedVersionRange} +import org.apache.kafka.common.utils.Utils + +import java.util.Properties +import scala.collection.Seq +import scala.collection.immutable.ListMap +import scala.jdk.CollectionConverters._ + +import joptsimple.OptionSpec + +object FeatureCommand { + + def main(args: Array[String]): Unit = { + val opts = new FeatureCommandOptions(args) + val featureApis = new FeatureApis(opts) + var exitCode = 0 + try { + featureApis.execute() + } catch { + case e: IllegalArgumentException => + printException(e) + opts.parser.printHelpOn(System.err) + exitCode = 1 + case e: Throwable => + printException(e) + exitCode = 1 + } finally { + featureApis.close() + Exit.exit(exitCode) + } + } + + private def printException(exception: Throwable): Unit = { + System.err.println("\nError encountered when executing command: " + Utils.stackTrace(exception)) + } +} + +class UpdateFeaturesException(message: String) extends RuntimeException(message) + +/** + * A class that provides necessary APIs to bridge the Admin client feature APIs with the CLI tool. + * + * @param opts the CLI options + */ +class FeatureApis(var opts: FeatureCommandOptions) { + private var supportedFeatures = BrokerFeatures.createDefault().supportedFeatures + private val adminClient = createAdminClient() + + private def pad(op: String): String = { + f"$op%11s" + } + + private val addOp = pad("[Add]") + private val upgradeOp = pad("[Upgrade]") + private val deleteOp = pad("[Delete]") + private val downgradeOp = pad("[Downgrade]") + + // For testing only. + def setSupportedFeatures(newFeatures: Features[SupportedVersionRange]): Unit = { + supportedFeatures = newFeatures + } + + // For testing only. + def setOptions(newOpts: FeatureCommandOptions): Unit = { + opts = newOpts + } + + private def describeFeatures(sendRequestToController: Boolean): FeatureMetadata = { + val options = new DescribeFeaturesOptions().sendRequestToController(sendRequestToController) + adminClient.describeFeatures(options).featureMetadata().get() + } + + /** + * Describes the supported and finalized features. If the --from-controller CLI option + * is provided, then the request is issued only to the controller, otherwise the request is issued + * to any of the provided bootstrap servers. + */ + def describeFeatures(): Unit = { + val result = describeFeatures(opts.hasFromControllerOption) + val features = result.supportedFeatures.asScala.keys.toSet ++ result.finalizedFeatures.asScala.keys.toSet + features.toList.sorted.foreach { Review comment: Why do we need to sort the features? ########## File path: core/src/main/scala/kafka/admin/FeatureCommand.scala ########## @@ -0,0 +1,350 @@ +/** + * 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.server.BrokerFeatures +import kafka.utils.{CommandDefaultOptions, CommandLineUtils, Exit} +import org.apache.kafka.clients.CommonClientConfigs +import org.apache.kafka.clients.admin.{Admin, DescribeFeaturesOptions, FeatureMetadata, FeatureUpdate, UpdateFeaturesOptions} +import org.apache.kafka.common.feature.{Features, SupportedVersionRange} +import org.apache.kafka.common.utils.Utils + +import java.util.Properties +import scala.collection.Seq +import scala.collection.immutable.ListMap +import scala.jdk.CollectionConverters._ + +import joptsimple.OptionSpec + +object FeatureCommand { + + def main(args: Array[String]): Unit = { + val opts = new FeatureCommandOptions(args) + val featureApis = new FeatureApis(opts) + var exitCode = 0 + try { + featureApis.execute() + } catch { + case e: IllegalArgumentException => + printException(e) + opts.parser.printHelpOn(System.err) + exitCode = 1 + case e: Throwable => + printException(e) + exitCode = 1 + } finally { + featureApis.close() + Exit.exit(exitCode) + } + } + + private def printException(exception: Throwable): Unit = { + System.err.println("\nError encountered when executing command: " + Utils.stackTrace(exception)) + } +} + +class UpdateFeaturesException(message: String) extends RuntimeException(message) + +/** + * A class that provides necessary APIs to bridge the Admin client feature APIs with the CLI tool. + * + * @param opts the CLI options + */ +class FeatureApis(var opts: FeatureCommandOptions) { + private var supportedFeatures = BrokerFeatures.createDefault().supportedFeatures + private val adminClient = createAdminClient() + + private def pad(op: String): String = { + f"$op%11s" + } + + private val addOp = pad("[Add]") + private val upgradeOp = pad("[Upgrade]") + private val deleteOp = pad("[Delete]") + private val downgradeOp = pad("[Downgrade]") + + // For testing only. + def setSupportedFeatures(newFeatures: Features[SupportedVersionRange]): Unit = { + supportedFeatures = newFeatures + } + + // For testing only. + def setOptions(newOpts: FeatureCommandOptions): Unit = { + opts = newOpts + } + + private def describeFeatures(sendRequestToController: Boolean): FeatureMetadata = { + val options = new DescribeFeaturesOptions().sendRequestToController(sendRequestToController) + adminClient.describeFeatures(options).featureMetadata().get() + } + + /** + * Describes the supported and finalized features. If the --from-controller CLI option + * is provided, then the request is issued only to the controller, otherwise the request is issued + * to any of the provided bootstrap servers. + */ + def describeFeatures(): Unit = { + val result = describeFeatures(opts.hasFromControllerOption) + val features = result.supportedFeatures.asScala.keys.toSet ++ result.finalizedFeatures.asScala.keys.toSet + features.toList.sorted.foreach { + case feature => + val (finalizedMinVersionLevel, finalizedMaxVersionLevel) = { + val finalizedVersionRange = result.finalizedFeatures.get(feature) + if (finalizedVersionRange == null) { + ("-", "-") + } else { + (finalizedVersionRange.minVersionLevel, finalizedVersionRange.maxVersionLevel) + } + } + val epoch = { + if (result.finalizedFeaturesEpoch.isPresent) { + result.finalizedFeaturesEpoch.get.toString + } else { + "-" + } + } + val (supportedMinVersion, supportedMaxVersion) = { + val supportedVersionRange = result.supportedFeatures.get(feature) + if (supportedVersionRange == null) { + ("-", "-") + } else { + (supportedVersionRange.minVersion, supportedVersionRange.maxVersion) + } + } + print(s"Feature: $feature") + print(s"\tSupportedMinVersion: $supportedMinVersion") + print(s"\tSupportedMaxVersion: $supportedMaxVersion") + print(s"\tFinalizedMinVersionLevel: $finalizedMinVersionLevel") + print(s"\tFinalizedMaxVersionLevel: $finalizedMaxVersionLevel") + println(s"\tEpoch: $epoch") + } + } + + /** + * Upgrades all features known to this tool to their highest max version levels. The method may + * add new finalized features if they were not finalized previously, but it does not delete + * any existing finalized feature. The results of the feature updates are written to STDOUT. + * + * NOTE: if the --dry-run CLI option is provided, this method only prints the expected feature + * updates to STDOUT, without applying them. + * + * @throws UpdateFeaturesException if at least one of the feature updates failed + */ + def upgradeAllFeatures(): Unit = { + val metadata = describeFeatures(true) + val existingFinalizedFeatures = metadata.finalizedFeatures + val updates = supportedFeatures.features.asScala.map { + case (feature, targetVersionRange) => + val existingVersionRange = existingFinalizedFeatures.get(feature) + if (existingVersionRange == null) { + val updateStr = + addOp + + s"\tFeature: $feature" + + s"\tExistingFinalizedMaxVersion: -" + + s"\tNewFinalizedMaxVersion: ${targetVersionRange.max}" + (feature, (updateStr, new FeatureUpdate(targetVersionRange.max, false))) + } else { + if (targetVersionRange.max > existingVersionRange.maxVersionLevel) { + val updateStr = + upgradeOp + + s"\tFeature: $feature" + + s"\tExistingFinalizedMaxVersion: ${existingVersionRange.maxVersionLevel}" + + s"\tNewFinalizedMaxVersion: ${targetVersionRange.max}" + (feature, (updateStr, new FeatureUpdate(targetVersionRange.max, false))) + } else { + (feature, null) + } + } + }.filter{ case(_, updateInfo) => updateInfo != null}.toMap + + if (updates.nonEmpty) { + maybeApplyFeatureUpdates(updates) + } + } + + /** + * Downgrades existing finalized features to the highest max version levels known to this tool. + * The method may delete existing finalized features if they are no longer seen to be supported, + * but it does not add a feature that was not finalized previously. The results of the feature + * updates are written to STDOUT. + * + * NOTE: if the --dry-run CLI option is provided, this method only prints the expected feature + * updates to STDOUT, without applying them. + * + * @throws UpdateFeaturesException if at least one of the feature updates failed + */ + def downgradeAllFeatures(): Unit = { + val metadata = describeFeatures(true) + val existingFinalizedFeatures = metadata.finalizedFeatures + val supportedFeaturesMap = supportedFeatures.features + val updates = existingFinalizedFeatures.asScala.map { + case (feature, existingVersionRange) => + val targetVersionRange = supportedFeaturesMap.get(feature) + if (targetVersionRange == null) { + val updateStr = + deleteOp + + s"\tFeature: $feature" + + s"\tExistingFinalizedMaxVersion: ${existingVersionRange.maxVersionLevel}" + + s"\tNewFinalizedMaxVersion: -" + (feature, (updateStr, new FeatureUpdate(0, true))) + } else { + if (targetVersionRange.max < existingVersionRange.maxVersionLevel) { + val updateStr = + downgradeOp + + s"\tFeature: $feature" + + s"\tExistingFinalizedMaxVersion: ${existingVersionRange.maxVersionLevel}" + + s"\tNewFinalizedMaxVersion: ${targetVersionRange.max}" + (feature, (updateStr, new FeatureUpdate(targetVersionRange.max, true))) + } else { + (feature, null) + } + } + }.filter{ case(_, updateInfo) => updateInfo != null}.toMap + + if (updates.nonEmpty) { + maybeApplyFeatureUpdates(updates) + } + } + + /** + * Applies the provided feature updates. If the --dry-run CLI option is provided, the method + * only prints the expected feature updates to STDOUT without applying them. + * + * @param updates the feature updates to be applied via the admin client + * + * @throws UpdateFeaturesException if at least one of the feature updates failed + */ + private def maybeApplyFeatureUpdates(updates: Map[String, (String, FeatureUpdate)]): Unit = { + if (opts.hasDryRunOption) { + println("Expected feature updates:") + println(ListMap(updates.toSeq.sortBy(_._1):_*) + .map { case(_, (updateStr, _)) => updateStr} + .mkString("\n")) + } else { + val result = adminClient.updateFeatures( + updates.map { case(feature, (_, update)) => (feature, update)}.asJava, + new UpdateFeaturesOptions()) + val failures = ListMap(result.values.asScala.toSeq.sortBy(_._1):_*).map { + case (feature, updateFuture) => + val (updateStr, _) = updates(feature) + try { + updateFuture.get + println(updateStr + "\tResult: OK") + 0 + } catch { + case e: Exception => + println(updateStr + "\tResult: FAILED due to " + e.getMessage) + 1 + } + }.sum + if (failures > 0) { + throw new UpdateFeaturesException(s"$failures feature updates failed!") + } + } + } + + def execute(): Unit = { + if (opts.hasDescribeOption) { + describeFeatures() + } else if (opts.hasUpgradeAllOption) { + upgradeAllFeatures() + } else if (opts.hasDowngradeAllOption) { + downgradeAllFeatures() + } else { + throw new IllegalStateException("Unexpected state: no CLI command could be executed.") + } + } + + def close(): Unit = { + adminClient.close() + } + + private def createAdminClient(): Admin = { + val props = new Properties() + props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, opts.bootstrapServers) + Admin.create(props) + } +} + +class FeatureCommandOptions(args: Array[String]) extends CommandDefaultOptions(args) { + private val bootstrapServerOpt = + parser.accepts("bootstrap-server", "REQUIRED: The Kafka server(s) to connect to.") + .withRequiredArg + .describedAs("server to connect to") + .ofType(classOf[String]) + private val describeOpt = parser.accepts( + "describe", + "Describe supported and finalized features. By default, the features are described from a" + + " random broker. The request can be optionally directed only to the controller using the" + + " --from-controller option.") + private val fromControllerOpt = parser.accepts( + "from-controller", + "Describe supported and finalized features from the controller.") + private val upgradeAllOpt = parser.accepts( + "upgrade-all", + "Upgrades all finalized features to the maximum version levels known to the tool." + + " This command finalizes new features known to the tool that were never finalized" + + " previously in the cluster, but it is guaranteed to not delete any existing feature.") + private val downgradeAllOpt = parser.accepts( + "downgrade-all", + "Downgrades all finalized features to the maximum version levels known to the tool." + + " This command deletes unknown features from the list of finalized features in the" + + " cluster, but it is guaranteed to not add a new feature.") + private val dryRunOpt = parser.accepts( + "dry-run", + "Performs a dry-run of upgrade/downgrade mutations to finalized feature without applying them.") + + options = parser.parse(args : _*) + + checkArgs() + + def has(builder: OptionSpec[_]): Boolean = options.has(builder) + + def hasDescribeOption: Boolean = has(describeOpt) + + def hasFromControllerOption: Boolean = has(fromControllerOpt) + + def hasDryRunOption: Boolean = has(dryRunOpt) + + def hasUpgradeAllOption: Boolean = has(upgradeAllOpt) + + def hasDowngradeAllOption: Boolean = has(downgradeAllOpt) + + def bootstrapServers: String = options.valueOf(bootstrapServerOpt) + + def checkArgs(): Unit = { + CommandLineUtils.printHelpAndExitIfNeeded(this, "This tool describes and updates finalized features.") + val numActions = Seq(describeOpt, upgradeAllOpt, downgradeAllOpt).count(has) + if (numActions != 1) { + CommandLineUtils.printUsageAndDie( + parser, + "Command must include exactly one action: --describe, --upgrade-all, --downgrade-all.") + } + CommandLineUtils.checkRequiredArgs(parser, options, bootstrapServerOpt) + if (hasDryRunOption && !hasUpgradeAllOption && !hasDowngradeAllOption) { + CommandLineUtils.printUsageAndDie( + parser, + "Command can contain --dry-run option only when either --upgrade-all or --downgrade-all actions are provided.") + } + if (hasFromControllerOption && !hasDescribeOption) { Review comment: Could we just log a warning in this case instead of failing? ########## File path: core/src/main/scala/kafka/admin/FeatureCommand.scala ########## @@ -0,0 +1,350 @@ +/** + * 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.server.BrokerFeatures +import kafka.utils.{CommandDefaultOptions, CommandLineUtils, Exit} +import org.apache.kafka.clients.CommonClientConfigs +import org.apache.kafka.clients.admin.{Admin, DescribeFeaturesOptions, FeatureMetadata, FeatureUpdate, UpdateFeaturesOptions} +import org.apache.kafka.common.feature.{Features, SupportedVersionRange} +import org.apache.kafka.common.utils.Utils + +import java.util.Properties +import scala.collection.Seq +import scala.collection.immutable.ListMap +import scala.jdk.CollectionConverters._ + +import joptsimple.OptionSpec + +object FeatureCommand { + + def main(args: Array[String]): Unit = { + val opts = new FeatureCommandOptions(args) + val featureApis = new FeatureApis(opts) + var exitCode = 0 + try { + featureApis.execute() + } catch { + case e: IllegalArgumentException => + printException(e) + opts.parser.printHelpOn(System.err) + exitCode = 1 + case e: Throwable => + printException(e) + exitCode = 1 + } finally { + featureApis.close() + Exit.exit(exitCode) + } + } + + private def printException(exception: Throwable): Unit = { + System.err.println("\nError encountered when executing command: " + Utils.stackTrace(exception)) + } +} + +class UpdateFeaturesException(message: String) extends RuntimeException(message) + +/** + * A class that provides necessary APIs to bridge the Admin client feature APIs with the CLI tool. + * + * @param opts the CLI options + */ +class FeatureApis(var opts: FeatureCommandOptions) { + private var supportedFeatures = BrokerFeatures.createDefault().supportedFeatures + private val adminClient = createAdminClient() + + private def pad(op: String): String = { + f"$op%11s" + } + + private val addOp = pad("[Add]") + private val upgradeOp = pad("[Upgrade]") + private val deleteOp = pad("[Delete]") + private val downgradeOp = pad("[Downgrade]") + + // For testing only. + def setSupportedFeatures(newFeatures: Features[SupportedVersionRange]): Unit = { + supportedFeatures = newFeatures + } + + // For testing only. + def setOptions(newOpts: FeatureCommandOptions): Unit = { + opts = newOpts + } + + private def describeFeatures(sendRequestToController: Boolean): FeatureMetadata = { + val options = new DescribeFeaturesOptions().sendRequestToController(sendRequestToController) + adminClient.describeFeatures(options).featureMetadata().get() + } + + /** + * Describes the supported and finalized features. If the --from-controller CLI option + * is provided, then the request is issued only to the controller, otherwise the request is issued + * to any of the provided bootstrap servers. + */ + def describeFeatures(): Unit = { + val result = describeFeatures(opts.hasFromControllerOption) + val features = result.supportedFeatures.asScala.keys.toSet ++ result.finalizedFeatures.asScala.keys.toSet + features.toList.sorted.foreach { + case feature => + val (finalizedMinVersionLevel, finalizedMaxVersionLevel) = { + val finalizedVersionRange = result.finalizedFeatures.get(feature) + if (finalizedVersionRange == null) { + ("-", "-") + } else { + (finalizedVersionRange.minVersionLevel, finalizedVersionRange.maxVersionLevel) + } + } + val epoch = { + if (result.finalizedFeaturesEpoch.isPresent) { + result.finalizedFeaturesEpoch.get.toString + } else { + "-" + } + } + val (supportedMinVersion, supportedMaxVersion) = { + val supportedVersionRange = result.supportedFeatures.get(feature) + if (supportedVersionRange == null) { + ("-", "-") + } else { + (supportedVersionRange.minVersion, supportedVersionRange.maxVersion) + } + } + print(s"Feature: $feature") + print(s"\tSupportedMinVersion: $supportedMinVersion") + print(s"\tSupportedMaxVersion: $supportedMaxVersion") + print(s"\tFinalizedMinVersionLevel: $finalizedMinVersionLevel") + print(s"\tFinalizedMaxVersionLevel: $finalizedMaxVersionLevel") + println(s"\tEpoch: $epoch") + } + } + + /** + * Upgrades all features known to this tool to their highest max version levels. The method may + * add new finalized features if they were not finalized previously, but it does not delete + * any existing finalized feature. The results of the feature updates are written to STDOUT. + * + * NOTE: if the --dry-run CLI option is provided, this method only prints the expected feature + * updates to STDOUT, without applying them. + * + * @throws UpdateFeaturesException if at least one of the feature updates failed + */ + def upgradeAllFeatures(): Unit = { + val metadata = describeFeatures(true) + val existingFinalizedFeatures = metadata.finalizedFeatures + val updates = supportedFeatures.features.asScala.map { + case (feature, targetVersionRange) => + val existingVersionRange = existingFinalizedFeatures.get(feature) + if (existingVersionRange == null) { + val updateStr = + addOp + + s"\tFeature: $feature" + + s"\tExistingFinalizedMaxVersion: -" + + s"\tNewFinalizedMaxVersion: ${targetVersionRange.max}" + (feature, (updateStr, new FeatureUpdate(targetVersionRange.max, false))) + } else { + if (targetVersionRange.max > existingVersionRange.maxVersionLevel) { + val updateStr = + upgradeOp + + s"\tFeature: $feature" + + s"\tExistingFinalizedMaxVersion: ${existingVersionRange.maxVersionLevel}" + + s"\tNewFinalizedMaxVersion: ${targetVersionRange.max}" + (feature, (updateStr, new FeatureUpdate(targetVersionRange.max, false))) + } else { + (feature, null) + } + } + }.filter{ case(_, updateInfo) => updateInfo != null}.toMap + + if (updates.nonEmpty) { + maybeApplyFeatureUpdates(updates) + } + } + + /** + * Downgrades existing finalized features to the highest max version levels known to this tool. + * The method may delete existing finalized features if they are no longer seen to be supported, + * but it does not add a feature that was not finalized previously. The results of the feature + * updates are written to STDOUT. + * + * NOTE: if the --dry-run CLI option is provided, this method only prints the expected feature + * updates to STDOUT, without applying them. + * + * @throws UpdateFeaturesException if at least one of the feature updates failed + */ + def downgradeAllFeatures(): Unit = { + val metadata = describeFeatures(true) + val existingFinalizedFeatures = metadata.finalizedFeatures + val supportedFeaturesMap = supportedFeatures.features + val updates = existingFinalizedFeatures.asScala.map { + case (feature, existingVersionRange) => + val targetVersionRange = supportedFeaturesMap.get(feature) + if (targetVersionRange == null) { + val updateStr = + deleteOp + + s"\tFeature: $feature" + + s"\tExistingFinalizedMaxVersion: ${existingVersionRange.maxVersionLevel}" + + s"\tNewFinalizedMaxVersion: -" + (feature, (updateStr, new FeatureUpdate(0, true))) + } else { + if (targetVersionRange.max < existingVersionRange.maxVersionLevel) { + val updateStr = + downgradeOp + + s"\tFeature: $feature" + + s"\tExistingFinalizedMaxVersion: ${existingVersionRange.maxVersionLevel}" + + s"\tNewFinalizedMaxVersion: ${targetVersionRange.max}" + (feature, (updateStr, new FeatureUpdate(targetVersionRange.max, true))) + } else { + (feature, null) + } + } + }.filter{ case(_, updateInfo) => updateInfo != null}.toMap + + if (updates.nonEmpty) { + maybeApplyFeatureUpdates(updates) + } + } + + /** + * Applies the provided feature updates. If the --dry-run CLI option is provided, the method + * only prints the expected feature updates to STDOUT without applying them. + * + * @param updates the feature updates to be applied via the admin client + * + * @throws UpdateFeaturesException if at least one of the feature updates failed + */ + private def maybeApplyFeatureUpdates(updates: Map[String, (String, FeatureUpdate)]): Unit = { + if (opts.hasDryRunOption) { + println("Expected feature updates:") + println(ListMap(updates.toSeq.sortBy(_._1):_*) + .map { case(_, (updateStr, _)) => updateStr} + .mkString("\n")) + } else { + val result = adminClient.updateFeatures( + updates.map { case(feature, (_, update)) => (feature, update)}.asJava, + new UpdateFeaturesOptions()) + val failures = ListMap(result.values.asScala.toSeq.sortBy(_._1):_*).map { + case (feature, updateFuture) => + val (updateStr, _) = updates(feature) + try { + updateFuture.get + println(updateStr + "\tResult: OK") + 0 + } catch { + case e: Exception => + println(updateStr + "\tResult: FAILED due to " + e.getMessage) + 1 + } + }.sum + if (failures > 0) { + throw new UpdateFeaturesException(s"$failures feature updates failed!") + } + } + } + + def execute(): Unit = { + if (opts.hasDescribeOption) { + describeFeatures() + } else if (opts.hasUpgradeAllOption) { + upgradeAllFeatures() + } else if (opts.hasDowngradeAllOption) { + downgradeAllFeatures() + } else { + throw new IllegalStateException("Unexpected state: no CLI command could be executed.") + } + } + + def close(): Unit = { + adminClient.close() + } + + private def createAdminClient(): Admin = { + val props = new Properties() + props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, opts.bootstrapServers) + Admin.create(props) + } +} + +class FeatureCommandOptions(args: Array[String]) extends CommandDefaultOptions(args) { + private val bootstrapServerOpt = + parser.accepts("bootstrap-server", "REQUIRED: The Kafka server(s) to connect to.") + .withRequiredArg + .describedAs("server to connect to") + .ofType(classOf[String]) + private val describeOpt = parser.accepts( + "describe", + "Describe supported and finalized features. By default, the features are described from a" + + " random broker. The request can be optionally directed only to the controller using the" + + " --from-controller option.") + private val fromControllerOpt = parser.accepts( + "from-controller", + "Describe supported and finalized features from the controller.") + private val upgradeAllOpt = parser.accepts( + "upgrade-all", + "Upgrades all finalized features to the maximum version levels known to the tool." + + " This command finalizes new features known to the tool that were never finalized" + + " previously in the cluster, but it is guaranteed to not delete any existing feature.") + private val downgradeAllOpt = parser.accepts( + "downgrade-all", + "Downgrades all finalized features to the maximum version levels known to the tool." + + " This command deletes unknown features from the list of finalized features in the" + + " cluster, but it is guaranteed to not add a new feature.") + private val dryRunOpt = parser.accepts( + "dry-run", + "Performs a dry-run of upgrade/downgrade mutations to finalized feature without applying them.") + + options = parser.parse(args : _*) + + checkArgs() + + def has(builder: OptionSpec[_]): Boolean = options.has(builder) + + def hasDescribeOption: Boolean = has(describeOpt) + + def hasFromControllerOption: Boolean = has(fromControllerOpt) + + def hasDryRunOption: Boolean = has(dryRunOpt) + + def hasUpgradeAllOption: Boolean = has(upgradeAllOpt) Review comment: A random thought I have is that since describe/upgrade/downgrade are mutually-exclusive, we could define one flag like `--action` and pass in the different options as strings to translate the an enum matching scenario. ########## File path: core/src/main/scala/kafka/admin/FeatureCommand.scala ########## @@ -0,0 +1,350 @@ +/** + * 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.server.BrokerFeatures +import kafka.utils.{CommandDefaultOptions, CommandLineUtils, Exit} +import org.apache.kafka.clients.CommonClientConfigs +import org.apache.kafka.clients.admin.{Admin, DescribeFeaturesOptions, FeatureMetadata, FeatureUpdate, UpdateFeaturesOptions} +import org.apache.kafka.common.feature.{Features, SupportedVersionRange} +import org.apache.kafka.common.utils.Utils + +import java.util.Properties +import scala.collection.Seq +import scala.collection.immutable.ListMap +import scala.jdk.CollectionConverters._ + +import joptsimple.OptionSpec + +object FeatureCommand { + + def main(args: Array[String]): Unit = { + val opts = new FeatureCommandOptions(args) + val featureApis = new FeatureApis(opts) + var exitCode = 0 + try { + featureApis.execute() + } catch { + case e: IllegalArgumentException => + printException(e) + opts.parser.printHelpOn(System.err) + exitCode = 1 + case e: Throwable => + printException(e) + exitCode = 1 + } finally { + featureApis.close() + Exit.exit(exitCode) + } + } + + private def printException(exception: Throwable): Unit = { + System.err.println("\nError encountered when executing command: " + Utils.stackTrace(exception)) + } +} + +class UpdateFeaturesException(message: String) extends RuntimeException(message) + +/** + * A class that provides necessary APIs to bridge the Admin client feature APIs with the CLI tool. + * + * @param opts the CLI options + */ +class FeatureApis(var opts: FeatureCommandOptions) { + private var supportedFeatures = BrokerFeatures.createDefault().supportedFeatures + private val adminClient = createAdminClient() + + private def pad(op: String): String = { + f"$op%11s" + } + + private val addOp = pad("[Add]") + private val upgradeOp = pad("[Upgrade]") + private val deleteOp = pad("[Delete]") + private val downgradeOp = pad("[Downgrade]") + + // For testing only. + def setSupportedFeatures(newFeatures: Features[SupportedVersionRange]): Unit = { + supportedFeatures = newFeatures + } + + // For testing only. + def setOptions(newOpts: FeatureCommandOptions): Unit = { + opts = newOpts + } + + private def describeFeatures(sendRequestToController: Boolean): FeatureMetadata = { + val options = new DescribeFeaturesOptions().sendRequestToController(sendRequestToController) + adminClient.describeFeatures(options).featureMetadata().get() + } + + /** + * Describes the supported and finalized features. If the --from-controller CLI option + * is provided, then the request is issued only to the controller, otherwise the request is issued + * to any of the provided bootstrap servers. + */ + def describeFeatures(): Unit = { + val result = describeFeatures(opts.hasFromControllerOption) + val features = result.supportedFeatures.asScala.keys.toSet ++ result.finalizedFeatures.asScala.keys.toSet + features.toList.sorted.foreach { + case feature => + val (finalizedMinVersionLevel, finalizedMaxVersionLevel) = { + val finalizedVersionRange = result.finalizedFeatures.get(feature) + if (finalizedVersionRange == null) { + ("-", "-") + } else { + (finalizedVersionRange.minVersionLevel, finalizedVersionRange.maxVersionLevel) + } + } + val epoch = { + if (result.finalizedFeaturesEpoch.isPresent) { + result.finalizedFeaturesEpoch.get.toString + } else { + "-" + } + } + val (supportedMinVersion, supportedMaxVersion) = { + val supportedVersionRange = result.supportedFeatures.get(feature) + if (supportedVersionRange == null) { + ("-", "-") + } else { + (supportedVersionRange.minVersion, supportedVersionRange.maxVersion) + } + } + print(s"Feature: $feature") + print(s"\tSupportedMinVersion: $supportedMinVersion") + print(s"\tSupportedMaxVersion: $supportedMaxVersion") + print(s"\tFinalizedMinVersionLevel: $finalizedMinVersionLevel") + print(s"\tFinalizedMaxVersionLevel: $finalizedMaxVersionLevel") + println(s"\tEpoch: $epoch") + } + } + + /** + * Upgrades all features known to this tool to their highest max version levels. The method may + * add new finalized features if they were not finalized previously, but it does not delete + * any existing finalized feature. The results of the feature updates are written to STDOUT. + * + * NOTE: if the --dry-run CLI option is provided, this method only prints the expected feature + * updates to STDOUT, without applying them. + * + * @throws UpdateFeaturesException if at least one of the feature updates failed + */ + def upgradeAllFeatures(): Unit = { + val metadata = describeFeatures(true) + val existingFinalizedFeatures = metadata.finalizedFeatures + val updates = supportedFeatures.features.asScala.map { + case (feature, targetVersionRange) => + val existingVersionRange = existingFinalizedFeatures.get(feature) + if (existingVersionRange == null) { + val updateStr = + addOp + + s"\tFeature: $feature" + + s"\tExistingFinalizedMaxVersion: -" + + s"\tNewFinalizedMaxVersion: ${targetVersionRange.max}" + (feature, (updateStr, new FeatureUpdate(targetVersionRange.max, false))) + } else { + if (targetVersionRange.max > existingVersionRange.maxVersionLevel) { + val updateStr = + upgradeOp + + s"\tFeature: $feature" + + s"\tExistingFinalizedMaxVersion: ${existingVersionRange.maxVersionLevel}" + + s"\tNewFinalizedMaxVersion: ${targetVersionRange.max}" + (feature, (updateStr, new FeatureUpdate(targetVersionRange.max, false))) + } else { + (feature, null) + } + } + }.filter{ case(_, updateInfo) => updateInfo != null}.toMap + + if (updates.nonEmpty) { + maybeApplyFeatureUpdates(updates) + } + } + + /** + * Downgrades existing finalized features to the highest max version levels known to this tool. + * The method may delete existing finalized features if they are no longer seen to be supported, + * but it does not add a feature that was not finalized previously. The results of the feature + * updates are written to STDOUT. + * + * NOTE: if the --dry-run CLI option is provided, this method only prints the expected feature + * updates to STDOUT, without applying them. + * + * @throws UpdateFeaturesException if at least one of the feature updates failed + */ + def downgradeAllFeatures(): Unit = { + val metadata = describeFeatures(true) + val existingFinalizedFeatures = metadata.finalizedFeatures + val supportedFeaturesMap = supportedFeatures.features + val updates = existingFinalizedFeatures.asScala.map { + case (feature, existingVersionRange) => + val targetVersionRange = supportedFeaturesMap.get(feature) + if (targetVersionRange == null) { + val updateStr = + deleteOp + + s"\tFeature: $feature" + + s"\tExistingFinalizedMaxVersion: ${existingVersionRange.maxVersionLevel}" + + s"\tNewFinalizedMaxVersion: -" + (feature, (updateStr, new FeatureUpdate(0, true))) + } else { + if (targetVersionRange.max < existingVersionRange.maxVersionLevel) { + val updateStr = + downgradeOp + + s"\tFeature: $feature" + + s"\tExistingFinalizedMaxVersion: ${existingVersionRange.maxVersionLevel}" + + s"\tNewFinalizedMaxVersion: ${targetVersionRange.max}" + (feature, (updateStr, new FeatureUpdate(targetVersionRange.max, true))) + } else { + (feature, null) + } + } + }.filter{ case(_, updateInfo) => updateInfo != null}.toMap + + if (updates.nonEmpty) { + maybeApplyFeatureUpdates(updates) + } + } + + /** + * Applies the provided feature updates. If the --dry-run CLI option is provided, the method + * only prints the expected feature updates to STDOUT without applying them. + * + * @param updates the feature updates to be applied via the admin client + * + * @throws UpdateFeaturesException if at least one of the feature updates failed + */ + private def maybeApplyFeatureUpdates(updates: Map[String, (String, FeatureUpdate)]): Unit = { + if (opts.hasDryRunOption) { + println("Expected feature updates:") + println(ListMap(updates.toSeq.sortBy(_._1):_*) + .map { case(_, (updateStr, _)) => updateStr} + .mkString("\n")) + } else { + val result = adminClient.updateFeatures( + updates.map { case(feature, (_, update)) => (feature, update)}.asJava, + new UpdateFeaturesOptions()) + val failures = ListMap(result.values.asScala.toSeq.sortBy(_._1):_*).map { + case (feature, updateFuture) => + val (updateStr, _) = updates(feature) + try { + updateFuture.get + println(updateStr + "\tResult: OK") + 0 + } catch { + case e: Exception => + println(updateStr + "\tResult: FAILED due to " + e.getMessage) + 1 + } + }.sum + if (failures > 0) { + throw new UpdateFeaturesException(s"$failures feature updates failed!") + } + } + } + + def execute(): Unit = { + if (opts.hasDescribeOption) { + describeFeatures() + } else if (opts.hasUpgradeAllOption) { + upgradeAllFeatures() + } else if (opts.hasDowngradeAllOption) { + downgradeAllFeatures() + } else { + throw new IllegalStateException("Unexpected state: no CLI command could be executed.") + } + } + + def close(): Unit = { + adminClient.close() + } + + private def createAdminClient(): Admin = { + val props = new Properties() + props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, opts.bootstrapServers) + Admin.create(props) + } +} + +class FeatureCommandOptions(args: Array[String]) extends CommandDefaultOptions(args) { + private val bootstrapServerOpt = + parser.accepts("bootstrap-server", "REQUIRED: The Kafka server(s) to connect to.") + .withRequiredArg + .describedAs("server to connect to") + .ofType(classOf[String]) + private val describeOpt = parser.accepts( + "describe", + "Describe supported and finalized features. By default, the features are described from a" + + " random broker. The request can be optionally directed only to the controller using the" + + " --from-controller option.") + private val fromControllerOpt = parser.accepts( + "from-controller", + "Describe supported and finalized features from the controller.") + private val upgradeAllOpt = parser.accepts( + "upgrade-all", + "Upgrades all finalized features to the maximum version levels known to the tool." + + " This command finalizes new features known to the tool that were never finalized" + + " previously in the cluster, but it is guaranteed to not delete any existing feature.") + private val downgradeAllOpt = parser.accepts( + "downgrade-all", + "Downgrades all finalized features to the maximum version levels known to the tool." + + " This command deletes unknown features from the list of finalized features in the" + + " cluster, but it is guaranteed to not add a new feature.") + private val dryRunOpt = parser.accepts( + "dry-run", + "Performs a dry-run of upgrade/downgrade mutations to finalized feature without applying them.") + + options = parser.parse(args : _*) + + checkArgs() + + def has(builder: OptionSpec[_]): Boolean = options.has(builder) + + def hasDescribeOption: Boolean = has(describeOpt) Review comment: I feel we don't need the `has` prefix for all the options, which seem not matching with the conventions by looking at examples as `CofnigCommand.scala` ########## File path: core/src/main/scala/kafka/admin/FeatureCommand.scala ########## @@ -0,0 +1,350 @@ +/** + * 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.server.BrokerFeatures +import kafka.utils.{CommandDefaultOptions, CommandLineUtils, Exit} +import org.apache.kafka.clients.CommonClientConfigs +import org.apache.kafka.clients.admin.{Admin, DescribeFeaturesOptions, FeatureMetadata, FeatureUpdate, UpdateFeaturesOptions} +import org.apache.kafka.common.feature.{Features, SupportedVersionRange} +import org.apache.kafka.common.utils.Utils + +import java.util.Properties +import scala.collection.Seq +import scala.collection.immutable.ListMap +import scala.jdk.CollectionConverters._ + +import joptsimple.OptionSpec + +object FeatureCommand { + + def main(args: Array[String]): Unit = { + val opts = new FeatureCommandOptions(args) + val featureApis = new FeatureApis(opts) + var exitCode = 0 + try { + featureApis.execute() + } catch { + case e: IllegalArgumentException => + printException(e) + opts.parser.printHelpOn(System.err) + exitCode = 1 + case e: Throwable => + printException(e) + exitCode = 1 + } finally { + featureApis.close() + Exit.exit(exitCode) + } + } + + private def printException(exception: Throwable): Unit = { + System.err.println("\nError encountered when executing command: " + Utils.stackTrace(exception)) + } +} + +class UpdateFeaturesException(message: String) extends RuntimeException(message) + +/** + * A class that provides necessary APIs to bridge the Admin client feature APIs with the CLI tool. + * + * @param opts the CLI options + */ +class FeatureApis(var opts: FeatureCommandOptions) { + private var supportedFeatures = BrokerFeatures.createDefault().supportedFeatures + private val adminClient = createAdminClient() + + private def pad(op: String): String = { + f"$op%11s" + } + + private val addOp = pad("[Add]") + private val upgradeOp = pad("[Upgrade]") + private val deleteOp = pad("[Delete]") + private val downgradeOp = pad("[Downgrade]") + + // For testing only. + def setSupportedFeatures(newFeatures: Features[SupportedVersionRange]): Unit = { + supportedFeatures = newFeatures + } + + // For testing only. + def setOptions(newOpts: FeatureCommandOptions): Unit = { + opts = newOpts + } + + private def describeFeatures(sendRequestToController: Boolean): FeatureMetadata = { + val options = new DescribeFeaturesOptions().sendRequestToController(sendRequestToController) + adminClient.describeFeatures(options).featureMetadata().get() + } + + /** + * Describes the supported and finalized features. If the --from-controller CLI option + * is provided, then the request is issued only to the controller, otherwise the request is issued + * to any of the provided bootstrap servers. + */ + def describeFeatures(): Unit = { + val result = describeFeatures(opts.hasFromControllerOption) + val features = result.supportedFeatures.asScala.keys.toSet ++ result.finalizedFeatures.asScala.keys.toSet + features.toList.sorted.foreach { + case feature => + val (finalizedMinVersionLevel, finalizedMaxVersionLevel) = { + val finalizedVersionRange = result.finalizedFeatures.get(feature) + if (finalizedVersionRange == null) { + ("-", "-") + } else { + (finalizedVersionRange.minVersionLevel, finalizedVersionRange.maxVersionLevel) + } + } + val epoch = { + if (result.finalizedFeaturesEpoch.isPresent) { + result.finalizedFeaturesEpoch.get.toString + } else { + "-" + } + } + val (supportedMinVersion, supportedMaxVersion) = { + val supportedVersionRange = result.supportedFeatures.get(feature) + if (supportedVersionRange == null) { + ("-", "-") + } else { + (supportedVersionRange.minVersion, supportedVersionRange.maxVersion) + } + } + print(s"Feature: $feature") + print(s"\tSupportedMinVersion: $supportedMinVersion") + print(s"\tSupportedMaxVersion: $supportedMaxVersion") + print(s"\tFinalizedMinVersionLevel: $finalizedMinVersionLevel") + print(s"\tFinalizedMaxVersionLevel: $finalizedMaxVersionLevel") + println(s"\tEpoch: $epoch") + } + } + + /** + * Upgrades all features known to this tool to their highest max version levels. The method may + * add new finalized features if they were not finalized previously, but it does not delete + * any existing finalized feature. The results of the feature updates are written to STDOUT. + * + * NOTE: if the --dry-run CLI option is provided, this method only prints the expected feature + * updates to STDOUT, without applying them. + * + * @throws UpdateFeaturesException if at least one of the feature updates failed + */ + def upgradeAllFeatures(): Unit = { + val metadata = describeFeatures(true) + val existingFinalizedFeatures = metadata.finalizedFeatures + val updates = supportedFeatures.features.asScala.map { + case (feature, targetVersionRange) => + val existingVersionRange = existingFinalizedFeatures.get(feature) + if (existingVersionRange == null) { + val updateStr = + addOp + + s"\tFeature: $feature" + + s"\tExistingFinalizedMaxVersion: -" + + s"\tNewFinalizedMaxVersion: ${targetVersionRange.max}" + (feature, (updateStr, new FeatureUpdate(targetVersionRange.max, false))) + } else { + if (targetVersionRange.max > existingVersionRange.maxVersionLevel) { + val updateStr = + upgradeOp + + s"\tFeature: $feature" + + s"\tExistingFinalizedMaxVersion: ${existingVersionRange.maxVersionLevel}" + + s"\tNewFinalizedMaxVersion: ${targetVersionRange.max}" + (feature, (updateStr, new FeatureUpdate(targetVersionRange.max, false))) + } else { + (feature, null) + } + } + }.filter{ case(_, updateInfo) => updateInfo != null}.toMap + + if (updates.nonEmpty) { + maybeApplyFeatureUpdates(updates) + } + } + + /** + * Downgrades existing finalized features to the highest max version levels known to this tool. + * The method may delete existing finalized features if they are no longer seen to be supported, + * but it does not add a feature that was not finalized previously. The results of the feature + * updates are written to STDOUT. + * + * NOTE: if the --dry-run CLI option is provided, this method only prints the expected feature + * updates to STDOUT, without applying them. + * + * @throws UpdateFeaturesException if at least one of the feature updates failed + */ + def downgradeAllFeatures(): Unit = { + val metadata = describeFeatures(true) + val existingFinalizedFeatures = metadata.finalizedFeatures + val supportedFeaturesMap = supportedFeatures.features + val updates = existingFinalizedFeatures.asScala.map { + case (feature, existingVersionRange) => + val targetVersionRange = supportedFeaturesMap.get(feature) + if (targetVersionRange == null) { + val updateStr = + deleteOp + + s"\tFeature: $feature" + + s"\tExistingFinalizedMaxVersion: ${existingVersionRange.maxVersionLevel}" + + s"\tNewFinalizedMaxVersion: -" + (feature, (updateStr, new FeatureUpdate(0, true))) + } else { + if (targetVersionRange.max < existingVersionRange.maxVersionLevel) { + val updateStr = + downgradeOp + + s"\tFeature: $feature" + + s"\tExistingFinalizedMaxVersion: ${existingVersionRange.maxVersionLevel}" + + s"\tNewFinalizedMaxVersion: ${targetVersionRange.max}" + (feature, (updateStr, new FeatureUpdate(targetVersionRange.max, true))) + } else { + (feature, null) + } + } + }.filter{ case(_, updateInfo) => updateInfo != null}.toMap + + if (updates.nonEmpty) { + maybeApplyFeatureUpdates(updates) + } + } + + /** + * Applies the provided feature updates. If the --dry-run CLI option is provided, the method + * only prints the expected feature updates to STDOUT without applying them. + * + * @param updates the feature updates to be applied via the admin client + * + * @throws UpdateFeaturesException if at least one of the feature updates failed + */ + private def maybeApplyFeatureUpdates(updates: Map[String, (String, FeatureUpdate)]): Unit = { + if (opts.hasDryRunOption) { + println("Expected feature updates:") + println(ListMap(updates.toSeq.sortBy(_._1):_*) + .map { case(_, (updateStr, _)) => updateStr} + .mkString("\n")) + } else { + val result = adminClient.updateFeatures( + updates.map { case(feature, (_, update)) => (feature, update)}.asJava, + new UpdateFeaturesOptions()) + val failures = ListMap(result.values.asScala.toSeq.sortBy(_._1):_*).map { + case (feature, updateFuture) => + val (updateStr, _) = updates(feature) + try { + updateFuture.get + println(updateStr + "\tResult: OK") + 0 + } catch { + case e: Exception => + println(updateStr + "\tResult: FAILED due to " + e.getMessage) + 1 + } + }.sum + if (failures > 0) { Review comment: I'm not sure the exception thrown here is necessary since we already printed out the result, is this just for testing purpose? ########## File path: core/src/test/scala/unit/kafka/admin/FeatureCommandTest.scala ########## @@ -0,0 +1,245 @@ +/** + * 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.api.KAFKA_2_7_IV0 +import kafka.server.{BaseRequestTest, KafkaConfig, KafkaServer} +import kafka.utils.TestUtils +import kafka.utils.TestUtils.waitUntilTrue +import org.apache.kafka.common.feature.{Features, SupportedVersionRange} +import org.apache.kafka.common.utils.Utils + +import java.util.Properties + +import org.junit.Assert.{assertEquals, assertTrue} +import org.junit.Test +import org.scalatest.Assertions.intercept + +class FeatureCommandTest extends BaseRequestTest { + override def brokerCount: Int = 3 + + override def brokerPropertyOverrides(props: Properties): Unit = { + props.put(KafkaConfig.InterBrokerProtocolVersionProp, KAFKA_2_7_IV0.toString) + } + + private def defaultSupportedFeatures(): Features[SupportedVersionRange] = { + Features.supportedFeatures(Utils.mkMap(Utils.mkEntry("feature_1", new SupportedVersionRange(1, 3)), + Utils.mkEntry("feature_2", new SupportedVersionRange(1, 5)))) + } + + private def updateSupportedFeatures(features: Features[SupportedVersionRange], + targetServers: Set[KafkaServer]): Unit = { + targetServers.foreach(s => { + s.brokerFeatures.setSupportedFeatures(features) + s.zkClient.updateBrokerInfo(s.createBrokerInfo) + }) + + // Wait until updates to all BrokerZNode supported features propagate to the controller. + val brokerIds = targetServers.map(s => s.config.brokerId) + waitUntilTrue( + () => servers.exists(s => { + if (s.kafkaController.isActive) { + s.kafkaController.controllerContext.liveOrShuttingDownBrokers + .filter(b => brokerIds.contains(b.id)) + .forall(b => { + b.features.equals(features) + }) + } else { + false + } + }), + "Controller did not get broker updates") + } + + private def updateSupportedFeaturesInAllBrokers(features: Features[SupportedVersionRange]): Unit = { + updateSupportedFeatures(features, Set[KafkaServer]() ++ servers) + } + + /** + * Tests if the FeatureApis#describeFeatures API works as expected when describing features before and + * after upgrading features. + */ + @Test + def testDescribeFeaturesSuccess(): Unit = { + updateSupportedFeaturesInAllBrokers(defaultSupportedFeatures()) + val featureApis = new FeatureApis(new FeatureCommandOptions(Array("--bootstrap-server", brokerList, "--describe", "--from-controller"))) + featureApis.setSupportedFeatures(defaultSupportedFeatures()) + try { + val initialDescribeOutput = TestUtils.grabConsoleOutput(featureApis.describeFeatures()) + val expectedInitialDescribeOutput = + "Feature: feature_1\tSupportedMinVersion: 1\tSupportedMaxVersion: 3\tFinalizedMinVersionLevel: -\tFinalizedMaxVersionLevel: -\tEpoch: 0\n" + + "Feature: feature_2\tSupportedMinVersion: 1\tSupportedMaxVersion: 5\tFinalizedMinVersionLevel: -\tFinalizedMaxVersionLevel: -\tEpoch: 0\n" + assertEquals(expectedInitialDescribeOutput, initialDescribeOutput) + featureApis.upgradeAllFeatures() + val finalDescribeOutput = TestUtils.grabConsoleOutput(featureApis.describeFeatures()) + val expectedFinalDescribeOutput = + "Feature: feature_1\tSupportedMinVersion: 1\tSupportedMaxVersion: 3\tFinalizedMinVersionLevel: 1\tFinalizedMaxVersionLevel: 3\tEpoch: 1\n" + + "Feature: feature_2\tSupportedMinVersion: 1\tSupportedMaxVersion: 5\tFinalizedMinVersionLevel: 1\tFinalizedMaxVersionLevel: 5\tEpoch: 1\n" + assertEquals(expectedFinalDescribeOutput, finalDescribeOutput) + } finally { + featureApis.close() + } + } + + /** + * Tests if the FeatureApis#upgradeAllFeatures API works as expected during a success case. + */ + @Test + def testUpgradeAllFeaturesSuccess(): Unit = { + val upgradeOpts = new FeatureCommandOptions(Array("--bootstrap-server", brokerList, "--upgrade-all")) + val featureApis = new FeatureApis(upgradeOpts) + try { + // Step (1): + // - Update the supported features across all brokers. + // - Upgrade non-existing feature_1 to maxVersionLevel: 2. + // - Verify results. + val initialSupportedFeatures = Features.supportedFeatures(Utils.mkMap(Utils.mkEntry("feature_1", new SupportedVersionRange(1, 2)))) + updateSupportedFeaturesInAllBrokers(initialSupportedFeatures) + featureApis.setSupportedFeatures(initialSupportedFeatures) + var output = TestUtils.grabConsoleOutput(featureApis.upgradeAllFeatures()) + var expected = + " [Add]\tFeature: feature_1\tExistingFinalizedMaxVersion: -\tNewFinalizedMaxVersion: 2\tResult: OK\n" + assertEquals(expected, output) + + // Step (2): + // - Update the supported features across all brokers. + // - Upgrade existing feature_1 to maxVersionLevel: 3. + // - Upgrade non-existing feature_2 to maxVersionLevel: 5. + // - Verify results. + updateSupportedFeaturesInAllBrokers(defaultSupportedFeatures()) + featureApis.setSupportedFeatures(defaultSupportedFeatures()) + output = TestUtils.grabConsoleOutput(featureApis.upgradeAllFeatures()) + expected = + " [Upgrade]\tFeature: feature_1\tExistingFinalizedMaxVersion: 2\tNewFinalizedMaxVersion: 3\tResult: OK\n" + + " [Add]\tFeature: feature_2\tExistingFinalizedMaxVersion: -\tNewFinalizedMaxVersion: 5\tResult: OK\n" + assertEquals(expected, output) + + // Step (3): + // - Perform an upgrade of all features again. + // - Since supported features have not changed, expect that the above action does not yield + // any results. + output = TestUtils.grabConsoleOutput(featureApis.upgradeAllFeatures()) + assertTrue(output.isEmpty) + featureApis.setOptions(upgradeOpts) + output = TestUtils.grabConsoleOutput(featureApis.upgradeAllFeatures()) + assertTrue(output.isEmpty) + } finally { + featureApis.close() + } + } + + /** + * Tests if the FeatureApis#downgradeAllFeatures API works as expected during a success case. + */ + @Test + def testDowngradeFeaturesSuccess(): Unit = { + val downgradeOpts = new FeatureCommandOptions(Array("--bootstrap-server", brokerList, "--downgrade-all")) + val upgradeOpts = new FeatureCommandOptions(Array("--bootstrap-server", brokerList, "--upgrade-all")) + val featureApis = new FeatureApis(upgradeOpts) + try { + // Step (1): + // - Update the supported features across all brokers. + // - Upgrade non-existing feature_1 to maxVersionLevel: 3. + // - Upgrade non-existing feature_2 to maxVersionLevel: 5. + updateSupportedFeaturesInAllBrokers(defaultSupportedFeatures()) + featureApis.setSupportedFeatures(defaultSupportedFeatures()) + featureApis.upgradeAllFeatures() + + // Step (2): + // - Downgrade existing feature_1 to maxVersionLevel: 2. + // - Delete feature_2 since it is no longer supported by the FeatureApis object. + // - Verify results. + val downgradedFeatures = Features.supportedFeatures(Utils.mkMap(Utils.mkEntry("feature_1", new SupportedVersionRange(1, 2)))) + featureApis.setSupportedFeatures(downgradedFeatures) + featureApis.setOptions(downgradeOpts) + var output = TestUtils.grabConsoleOutput(featureApis.downgradeAllFeatures()) + var expected = + "[Downgrade]\tFeature: feature_1\tExistingFinalizedMaxVersion: 3\tNewFinalizedMaxVersion: 2\tResult: OK\n" + + " [Delete]\tFeature: feature_2\tExistingFinalizedMaxVersion: 5\tNewFinalizedMaxVersion: -\tResult: OK\n" + assertEquals(expected, output) + + // Step (3): + // - Perform a downgrade of all features again. + // - Since supported features have not changed, expect that the above action does not yield + // any results. + updateSupportedFeaturesInAllBrokers(downgradedFeatures) + output = TestUtils.grabConsoleOutput(featureApis.downgradeAllFeatures()) + assertTrue(output.isEmpty) + + // Step (4): + // - Delete feature_1 since it is no longer supported by the FeatureApis object. + // - Verify results. + featureApis.setSupportedFeatures(Features.emptySupportedFeatures()) + output = TestUtils.grabConsoleOutput(featureApis.downgradeAllFeatures()) + expected = + " [Delete]\tFeature: feature_1\tExistingFinalizedMaxVersion: 2\tNewFinalizedMaxVersion: -\tResult: OK\n" + assertEquals(expected, output) + } finally { + featureApis.close() + } + } + + /** + * Tests if the FeatureApis#upgradeAllFeatures API works as expected during a partial failure case. + */ + @Test + def testUpgradeFeaturesFailure(): Unit = { Review comment: Do we need a downgrade failure test as well? ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
