<commit_before>/**
 * 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.tools.TerseFailure
import kafka.utils.Exit
import net.sourceforge.argparse4j.ArgumentParsers
import net.sourceforge.argparse4j.impl.Arguments.{append, fileType, store, storeTrue}
import net.sourceforge.argparse4j.inf.{ArgumentParserException, Namespace, Subparsers}
import net.sourceforge.argparse4j.internal.HelpScreenException
import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.clients.admin.FeatureUpdate.UpgradeType
import org.apache.kafka.clients.admin.{Admin, FeatureUpdate, UpdateFeaturesOptions}
import org.apache.kafka.common.utils.Utils
import org.apache.kafka.server.common.MetadataVersion

import java.io.{File, PrintStream}
import java.util.Properties
import scala.concurrent.ExecutionException
import scala.jdk.CollectionConverters._
import scala.compat.java8.OptionConverters._

object FeatureCommand {
  def main(args: Array[String]): Unit = {
    val res = mainNoExit(args, System.out)
    Exit.exit(res)
  }

  // This is used for integration tests in order to avoid killing the test with Exit.exit,
  // and in order to capture the command output.
  def mainNoExit(
    args: Array[String],
    out: PrintStream
  ): Int = {
    val parser = ArgumentParsers.newArgumentParser("kafka-features")
      .defaultHelp(true)
      .description("This tool manages feature flags in Kafka.")
    parser.addArgument("--bootstrap-server")
      .help("A comma-separated list of host:port pairs to use for establishing the connection to the Kafka cluster.")
      .required(true)

    parser.addArgument("--command-config")
      .`type`(fileType())
      .help("Property file containing configs to be passed to Admin Client.")
    val subparsers = parser.addSubparsers().dest("command")
    addDescribeParser(subparsers)
    addUpgradeParser(subparsers)
    addDowngradeParser(subparsers)
    addDisableParser(subparsers)

    try {
      val namespace = parser.parseArgs(args)
      val command = namespace.getString("command")

      val commandConfig = namespace.get[File]("command_config")
      val props = if (commandConfig != null) {
        if (!commandConfig.exists()) {
          throw new TerseFailure(s"Properties file ${commandConfig.getPath} does not exists!")
        }
        Utils.loadProps(commandConfig.getPath)
      } else {
        new Properties()
      }
      props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, namespace.getString("bootstrap_server"))
      val admin = Admin.create(props)
      try {
        command match {
          case "describe" => handleDescribe(out, admin)
          case "upgrade" => handleUpgrade(out, namespace, admin)
          case "downgrade" => handleDowngrade(out, namespace, admin)
          case "disable" => handleDisable(out, namespace, admin)
        }
      } finally {
        admin.close()
      }
      0
    } catch {
      case _: HelpScreenException =>
        0
      case e: ArgumentParserException =>
        System.err.println(s"Command line error: ${e.getMessage}. Type --help for help.")
        1
      case e: TerseFailure =>
        System.err.println(e.getMessage)
        1
    }
  }

  def addDescribeParser(subparsers: Subparsers): Unit = {
    subparsers.addParser("describe")
      .help("Describes the current active feature flags.")
  }

  def addUpgradeParser(subparsers: Subparsers): Unit = {
    val upgradeParser = subparsers.addParser("upgrade")
      .help("Upgrade one or more feature flags.")
    upgradeParser.addArgument("--metadata")
      .help("The level to which we should upgrade the metadata. For example, 3.3-IV3.")
      .action(store())
    upgradeParser.addArgument("--feature")
      .help("A feature upgrade we should perform, in key=value format. For example metadata.version=3.3-IV3.")
      .action(append())
    upgradeParser.addArgument("--dry-run")
      .help("Validate this upgrade, but do not perform it.")
      .action(storeTrue())
  }

  def addDowngradeParser(subparsers: Subparsers): Unit = {
    val downgradeParser = subparsers.addParser("downgrade")
      .help("Downgrade one or more feature flags.")
    downgradeParser.addArgument("--metadata")
      .help("The level to which we should downgrade the metadata. For example, 3.3-IV0.")
      .action(store())
    downgradeParser.addArgument("--feature")
      .help("A feature downgrade we should perform, in key=value format. " +
        "For example metadata.version=3.3-IV0.")
      .action(append())
    downgradeParser.addArgument("--unsafe")
      .help("Perform this downgrade even if it may irreversibly destroy metadata.")
      .action(storeTrue())
    downgradeParser.addArgument("--dry-run")
      .help("Validate this downgrade, but do not perform it.")
      .action(storeTrue())
  }

  def addDisableParser(subparsers: Subparsers): Unit = {
    val disableParser = subparsers.addParser("disable")
      .help("Disable one or more feature flags. This is the same as downgrading the version to zero.")
    disableParser.addArgument("--feature")
      .help("A feature flag to disable.")
      .action(append())
    disableParser.addArgument("--unsafe")
      .help("Disable this feature flag even if it may irreversibly destroy metadata.")
      .action(storeTrue())
    disableParser.addArgument("--dry-run")
      .help("Perform a dry-run of this disable operation.")
      .action(storeTrue())
  }

  def levelToString(
    feature: String,<commit_msg>MINOR: Fix incorrect example in feature command help (#12732)<commit_after>/**
 * 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.tools.TerseFailure
import kafka.utils.Exit
import net.sourceforge.argparse4j.ArgumentParsers
import net.sourceforge.argparse4j.impl.Arguments.{append, fileType, store, storeTrue}
import net.sourceforge.argparse4j.inf.{ArgumentParserException, Namespace, Subparsers}
import net.sourceforge.argparse4j.internal.HelpScreenException
import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.clients.admin.FeatureUpdate.UpgradeType
import org.apache.kafka.clients.admin.{Admin, FeatureUpdate, UpdateFeaturesOptions}
import org.apache.kafka.common.utils.Utils
import org.apache.kafka.server.common.MetadataVersion

import java.io.{File, PrintStream}
import java.util.Properties
import scala.concurrent.ExecutionException
import scala.jdk.CollectionConverters._
import scala.compat.java8.OptionConverters._

object FeatureCommand {
  def main(args: Array[String]): Unit = {
    val res = mainNoExit(args, System.out)
    Exit.exit(res)
  }

  // This is used for integration tests in order to avoid killing the test with Exit.exit,
  // and in order to capture the command output.
  def mainNoExit(
    args: Array[String],
    out: PrintStream
  ): Int = {
    val parser = ArgumentParsers.newArgumentParser("kafka-features")
      .defaultHelp(true)
      .description("This tool manages feature flags in Kafka.")
    parser.addArgument("--bootstrap-server")
      .help("A comma-separated list of host:port pairs to use for establishing the connection to the Kafka cluster.")
      .required(true)

    parser.addArgument("--command-config")
      .`type`(fileType())
      .help("Property file containing configs to be passed to Admin Client.")
    val subparsers = parser.addSubparsers().dest("command")
    addDescribeParser(subparsers)
    addUpgradeParser(subparsers)
    addDowngradeParser(subparsers)
    addDisableParser(subparsers)

    try {
      val namespace = parser.parseArgs(args)
      val command = namespace.getString("command")

      val commandConfig = namespace.get[File]("command_config")
      val props = if (commandConfig != null) {
        if (!commandConfig.exists()) {
          throw new TerseFailure(s"Properties file ${commandConfig.getPath} does not exists!")
        }
        Utils.loadProps(commandConfig.getPath)
      } else {
        new Properties()
      }
      props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, namespace.getString("bootstrap_server"))
      val admin = Admin.create(props)
      try {
        command match {
          case "describe" => handleDescribe(out, admin)
          case "upgrade" => handleUpgrade(out, namespace, admin)
          case "downgrade" => handleDowngrade(out, namespace, admin)
          case "disable" => handleDisable(out, namespace, admin)
        }
      } finally {
        admin.close()
      }
      0
    } catch {
      case _: HelpScreenException =>
        0
      case e: ArgumentParserException =>
        System.err.println(s"Command line error: ${e.getMessage}. Type --help for help.")
        1
      case e: TerseFailure =>
        System.err.println(e.getMessage)
        1
    }
  }

  def addDescribeParser(subparsers: Subparsers): Unit = {
    subparsers.addParser("describe")
      .help("Describes the current active feature flags.")
  }

  def addUpgradeParser(subparsers: Subparsers): Unit = {
    val upgradeParser = subparsers.addParser("upgrade")
      .help("Upgrade one or more feature flags.")
    upgradeParser.addArgument("--metadata")
      .help("The level to which we should upgrade the metadata. For example, 3.3-IV3.")
      .action(store())
    upgradeParser.addArgument("--feature")
      .help("A feature upgrade we should perform, in feature=level format. For example: `metadata.version=5`.")
      .action(append())
    upgradeParser.addArgument("--dry-run")
      .help("Validate this upgrade, but do not perform it.")
      .action(storeTrue())
  }

  def addDowngradeParser(subparsers: Subparsers): Unit = {
    val downgradeParser = subparsers.addParser("downgrade")
      .help("Downgrade one or more feature flags.")
    downgradeParser.addArgument("--metadata")
      .help("The level to which we should downgrade the metadata. For example, 3.3-IV0.")
      .action(store())
    downgradeParser.addArgument("--feature")
      .help("A feature downgrade we should perform, in feature=level format. For example: `metadata.version=5`.")
      .action(append())
    downgradeParser.addArgument("--unsafe")
      .help("Perform this downgrade even if it may irreversibly destroy metadata.")
      .action(storeTrue())
    downgradeParser.addArgument("--dry-run")
      .help("Validate this downgrade, but do not perform it.")
      .action(storeTrue())
  }

  def addDisableParser(subparsers: Subparsers): Unit = {
    val disableParser = subparsers.addParser("disable")
      .help("Disable one or more feature flags. This is the same as downgrading the version to zero.")
    disableParser.addArgument("--feature")
      .help("A feature flag to disable.")
      .action(append())
    disableParser.addArgument("--unsafe")
      .help("Disable this feature flag even if it may irreversibly destroy metadata.")
      .action(storeTrue())
    disableParser.addArgument("--dry-run")
      .help("Perform a dry-run of this disable operation.")
      .action(storeTrue())
  }

  def levelToString(
    feature: String,
{'fraction_non_alphanumeric': 0.10590680201422747, 'fraction_numerical': 0.004076412756774039, 'mean_word_length': 3.457427858924118, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5558271', 'n_tokens_mistral': 3487, 'n_tokens_neox': 3429, 'n_words': 1017}