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

Code before:
    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("Unknown metadata.version " + releaseVersion +
          ". Supported metadata.version are " + 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]]))
    configToLogDirectories(config).foreach(formatter.addDirectory(_))

Review comment (bug):
Could we use an interpolated string here?
```suggestion
          throw new TerseFailure(s"Unknown metadata.version $releaseVersion. Supported metadata.version are ${
            metadataVersionsToString(
              MetadataVersion.MINIMUM_VERSION, MetadataVersion.latestProduction())
          }")
```

Code after:
    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]]))
    configToLogDirectories(config).foreach(formatter.addDirectory(_))
    formatter.run()