Repository: apache/kafka
File: core/src/main/scala/kafka/tools/StorageTool.scala (Scala)

Code before:
    printStream: PrintStream
  ): Unit = {
    if (config.processRoles.isEmpty) {
      throw new TerseFailure("The kafka configuration file appears to be for " +
        "a legacy cluster. Formatting is only supported for clusters in KRaft mode.")
    }
    val formatter = new Formatter().
      setPrintStream(printStream).
      setNodeId(config.nodeId).
      setClusterId(namespace.getString("cluster_id")).
      setUnstableFeatureVersionsEnabled(config.unstableFeatureVersionsEnabled).
      setIgnoreFormatted(namespace.getBoolean("ignore_formatted")).
      setControllerListenerName(config.controllerListenerNames.get(0)).
      setMetadataLogDirectory(config.metadataLogDir)
    Option(namespace.getString("release_version")).foreach(
      releaseVersion => formatter.
        setReleaseVersion(MetadataVersion.fromVersionString(releaseVersion)))
    Option(namespace.getList[String]("feature")).foreach(
      featureNamesAndLevels(_).foreachEntry {
        (k, v) => formatter.setFeatureLevel(k, v)
      })
    Option(namespace.getString("initial_controllers")).
      foreach(v => formatter.setInitialControllers(DynamicVoters.parse(v)))
    if (namespace.getBoolean("standalone")) {
      formatter.setInitialControllers(createStandaloneDynamicVoters(config))
    }
    if (namespace.getBoolean("no_initial_controllers")) {
      formatter.setNoInitialControllersFlag(true)
    } else {
      if (config.processRoles.contains(ProcessRole.ControllerRole)) {
        if (config.quorumConfig.voters().isEmpty && formatter.initialVoters().isEmpty) {
          throw new TerseFailure("Because " + QuorumConfig.QUORUM_VOTERS_CONFIG +
            " is not set on this controller, you must specify one of the following: " +
            "--standalone, --initial-controllers, or --no-initial-controllers.");
        }
      }
    }
    Option(namespace.getList("add_scram")).
      foreach(scramArgs => formatter.setScramArguments(scramArgs.asInstanceOf[util.List[String]]))
    configToLogDirectories(config).foreach(formatter.addDirectory(_))
    formatter.run()
  }

  /**
   * Maps the given release version to the corresponding metadata version
   * and prints the corresponding features.
   *
   * @param namespace       Arguments containing the release version.
   * @param printStream     The print stream to output the version mapping.
   * @param validFeatures   List of features to be considered in the output
   */

Review comment (suggestion):
This is not very accurate. If the internal config unstable.feature.versions.enable is set, we actually allow non-production MVs. All callers of `MetadataVersion.fromVersionString` are either on the server side or from the tools and should have access to the unstable.feature.versions.enable. Perhaps it's better to change `MetadataVersion.fromVersionString` to take an enableUnstableFeature flag.



Code after:
    printStream: PrintStream
  ): Unit = {
    if (config.processRoles.isEmpty) {
      throw new TerseFailure("The kafka configuration file appears to be for " +
        "a legacy cluster. Formatting is only supported for clusters in KRaft mode.")
    }
    val formatter = new Formatter().
      setPrintStream(printStream).
      setNodeId(config.nodeId).
      setClusterId(namespace.getString("cluster_id")).
      setUnstableFeatureVersionsEnabled(config.unstableFeatureVersionsEnabled).
      setIgnoreFormatted(namespace.getBoolean("ignore_formatted")).
      setControllerListenerName(config.controllerListenerNames.get(0)).
      setMetadataLogDirectory(config.metadataLogDir)

    def metadataVersionsToString(first: MetadataVersion, last: MetadataVersion): String = {
      val versions = MetadataVersion.VERSIONS.slice(first.ordinal, last.ordinal + 1)
      versions.map(_.toString).mkString(", ")
    }
    Option(namespace.getString("release_version")).foreach(releaseVersion => {
      try {
        formatter.setReleaseVersion(MetadataVersion.fromVersionString(releaseVersion))
      } catch {
        case _: Throwable =>
          throw new TerseFailure(s"Unknown metadata.version $releaseVersion. Supported metadata.version are " +
            s"${metadataVersionsToString(MetadataVersion.MINIMUM_VERSION, MetadataVersion.latestProduction())}")
      }
    })

    Option(namespace.getList[String]("feature")).foreach(
      featureNamesAndLevels(_).foreachEntry {
        (k, v) => formatter.setFeatureLevel(k, v)
      })
    Option(namespace.getString("initial_controllers")).
      foreach(v => formatter.setInitialControllers(DynamicVoters.parse(v)))
    if (namespace.getBoolean("standalone")) {
      formatter.setInitialControllers(createStandaloneDynamicVoters(config))
    }
    if (namespace.getBoolean("no_initial_controllers")) {
      formatter.setNoInitialControllersFlag(true)
    } else {
      if (config.processRoles.contains(ProcessRole.ControllerRole)) {
        if (config.quorumConfig.voters().isEmpty && formatter.initialVoters().isEmpty) {
          throw new TerseFailure("Because " + QuorumConfig.QUORUM_VOTERS_CONFIG +
            " is not set on this controller, you must specify one of the following: " +
            "--standalone, --initial-controllers, or --no-initial-controllers.");
        }
      }
    }
    Option(namespace.getList("add_scram")).
      foreach(scramArgs => formatter.setScramArguments(scramArgs.asInstanceOf[util.List[String]]))