Repository: apache/kafka
File: core/src/main/scala/kafka/server/ConfigHelper.scala (Scala)

Code before:

  private def buildDescribeConfigsResult(resource: DescribeConfigsResource,
                                         configs: java.util.stream.Stream[java.util.Map.Entry[String, String]],
                                         createConfigEntry: (String, Any) => DescribeConfigsResponseData.DescribeConfigsResourceResult): DescribeConfigsResponseData.DescribeConfigsResult = {
    val configEntries = configs
      .filter(entry =>
        resource.configurationKeys == null ||
          resource.configurationKeys.isEmpty ||
          resource.configurationKeys.contains(entry.getKey)
      )
      .map[DescribeConfigsResponseData.DescribeConfigsResourceResult](entry => createConfigEntry(entry.getKey, entry.getValue))
      .toList

    new DescribeConfigsResponseData.DescribeConfigsResult()
      .setErrorCode(Errors.NONE.code)
      .setConfigs(configEntries)
  }

  private def createResponseConfig(resource: DescribeConfigsResource, configs: AbstractConfig,
                           createConfigEntry: (String, Any) => DescribeConfigsResponseData.DescribeConfigsResourceResult): DescribeConfigsResponseData.DescribeConfigsResult = {
    val nonInternalValues = configs.nonInternalValues // cache to avoid multiple calls
    val nonInternalValuesStream = nonInternalValues.entrySet().stream()
      .map(e => new java.util.AbstractMap.SimpleEntry[String, String](e.getKey, if (e.getValue == null) null else e.getValue.toString))
    val originalsFilteredStream = configs.originals.entrySet.stream()
      .filter(e => e.getValue != null && !nonInternalValues.containsKey(e.getKey)) // skip keys in nonInternalValues
      .map(e => new java.util.AbstractMap.SimpleEntry[String, String](e.getKey, if (e.getValue == null) null else e.getValue.toString))
    buildDescribeConfigsResult(resource, java.util.stream.Stream.concat(nonInternalValuesStream, originalsFilteredStream), createConfigEntry)
  }

  private def createResponseConfig(resource: DescribeConfigsResource, configs: java.util.Map[String, String],
                           createConfigEntry: (String, Any) => DescribeConfigsResponseData.DescribeConfigsResourceResult): DescribeConfigsResponseData.DescribeConfigsResult = {
    buildDescribeConfigsResult(resource, configs.entrySet().stream(), createConfigEntry)
  }

  def describeConfigs(resourceToConfigNames: List[DescribeConfigsResource],
                      includeSynonyms: Boolean,
                      includeDocumentation: Boolean): List[DescribeConfigsResponseData.DescribeConfigsResult] = {
    resourceToConfigNames.map { resource =>
      try {
        val configResult = ConfigResource.Type.forId(resource.resourceType) match {
          case ConfigResource.Type.TOPIC =>
            val topic = resource.resourceName
            Topic.validate(topic)
            if (metadataCache.contains(topic)) {
              val topicProps = configRepository.topicConfig(topic)
              val logConfig = LogConfig.fromProps(config.extractLogConfigMap, topicProps)
              createResponseConfig(resource, logConfig, createTopicConfigEntry(logConfig, topicProps, includeSynonyms, includeDocumentation))
            } else {
              new DescribeConfigsResponseData.DescribeConfigsResult().setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code)
                .setConfigs(Collections.emptyList[DescribeConfigsResponseData.DescribeConfigsResourceResult])
            }

Review comment (refactor):
It seems you are in trouble by the type conversion. Perhaps a better way is you could migrate those helper methods to server module by rewriting them to java code. WDYT?

Code after:
    new DescribeConfigsResponseData().setResults((authorizedConfigs ++ unauthorizedConfigs).asJava)
  }

  def describeConfigs(resourceToConfigNames: List[DescribeConfigsResource],
                      includeSynonyms: Boolean,
                      includeDocumentation: Boolean): List[DescribeConfigsResponseData.DescribeConfigsResult] = {
    resourceToConfigNames.map { resource =>
      try {
        val configResult = ConfigResource.Type.forId(resource.resourceType) match {
          case ConfigResource.Type.TOPIC =>
            val topic = resource.resourceName
            Topic.validate(topic)
            if (metadataCache.contains(topic)) {
              val topicProps = configRepository.topicConfig(topic)
              val logConfig = LogConfig.fromProps(config.extractLogConfigMap, topicProps)
              createResponseConfig(resource, logConfig, createTopicConfigEntry(logConfig, topicProps, includeSynonyms, includeDocumentation)(_, _))
            } else {
              new DescribeConfigsResponseData.DescribeConfigsResult().setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code)
                .setConfigs(Collections.emptyList[DescribeConfigsResponseData.DescribeConfigsResourceResult])
            }

          case ConfigResource.Type.BROKER =>
            if (resource.resourceName == null || resource.resourceName.isEmpty)
              createResponseConfig(resource, config.dynamicConfig.currentDynamicDefaultConfigs.asJava,
                createBrokerConfigEntry(perBrokerConfig = false, includeSynonyms, includeDocumentation)(_, _))
            else if (resourceNameToBrokerId(resource.resourceName) == config.brokerId)
              createResponseConfig(resource, config,
                createBrokerConfigEntry(perBrokerConfig = true, includeSynonyms, includeDocumentation)(_, _))
            else
              throw new InvalidRequestException(s"Unexpected broker id, expected ${config.brokerId} or empty string, but received ${resource.resourceName}")

          case ConfigResource.Type.BROKER_LOGGER =>
            if (resource.resourceName == null || resource.resourceName.isEmpty)
              throw new InvalidRequestException("Broker id must not be empty")
            else if (resourceNameToBrokerId(resource.resourceName) != config.brokerId)
              throw new InvalidRequestException(s"Unexpected broker id, expected ${config.brokerId} but received ${resource.resourceName}")
            else
              createResponseConfig(resource, LoggingController.loggers,
                (name: String, value: Object) => new DescribeConfigsResponseData.DescribeConfigsResourceResult().setName(name)
                  .setValue(value.toString).setConfigSource(ConfigSource.DYNAMIC_BROKER_LOGGER_CONFIG.id)
                  .setIsSensitive(false).setReadOnly(false).setSynonyms(List.empty.asJava))

          case ConfigResource.Type.CLIENT_METRICS =>
            if (resource.resourceName == null || resource.resourceName.isEmpty) {
              throw new InvalidRequestException("Client metrics subscription name must not be empty")
            } else {
              val clientMetricsProps = configRepository.config(new ConfigResource(ConfigResource.Type.CLIENT_METRICS, resource.resourceName))
              val clientMetricsConfig = ClientMetricsConfigs.fromProps(ClientMetricsConfigs.defaultConfigsMap(), clientMetricsProps)
              createResponseConfig(resource, clientMetricsConfig, createClientMetricsConfigEntry(clientMetricsConfig, clientMetricsProps, includeSynonyms, includeDocumentation)(_, _))
            }