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

Code before:
          authHelper.authorize(request.context, DESCRIBE_CONFIGS, GROUP, resource.resourceName)
        case rt => throw new InvalidRequestException(s"Unexpected resource type $rt for resource ${resource.resourceName}")
      }
    }
    val authorizedConfigs = describeConfigs(authorizedResources.toList, describeConfigsRequest.data.includeSynonyms, describeConfigsRequest.data.includeDocumentation)
    val unauthorizedConfigs = unauthorizedResources.map { resource =>
      val error = ConfigResource.Type.forId(resource.resourceType) match {
        case ConfigResource.Type.BROKER | ConfigResource.Type.BROKER_LOGGER | ConfigResource.Type.CLIENT_METRICS => Errors.CLUSTER_AUTHORIZATION_FAILED
        case ConfigResource.Type.TOPIC => Errors.TOPIC_AUTHORIZATION_FAILED
        case ConfigResource.Type.GROUP => Errors.GROUP_AUTHORIZATION_FAILED
        case rt => throw new InvalidRequestException(s"Unexpected resource type $rt for resource ${resource.resourceName}")
      }
      new DescribeConfigsResponseData.DescribeConfigsResult().setErrorCode(error.code)
        .setErrorMessage(error.message)
        .setConfigs(Collections.emptyList[DescribeConfigsResponseData.DescribeConfigsResourceResult])
        .setResourceName(resource.resourceName)
        .setResourceType(resource.resourceType)
    }
    new DescribeConfigsResponseData().setResults((authorizedConfigs ++ unauthorizedConfigs).asJava)
  }

  def describeConfigs(resourceToConfigNames: List[DescribeConfigsResource],
                      includeSynonyms: Boolean,
                      includeDocumentation: Boolean): List[DescribeConfigsResponseData.DescribeConfigsResult] = {
    resourceToConfigNames.map { resource =>
      def createResponseConfig(configs: Any,
                               createConfigEntry: (String, Any) => DescribeConfigsResponseData.DescribeConfigsResourceResult): DescribeConfigsResponseData.DescribeConfigsResult = {
        val stream = configs match {
          case c: AbstractConfig =>
            val nonInternalValues = c.nonInternalValues
            val originalsFiltered = c.originals.entrySet.stream()
              .filter(e =>
                e.getValue != null &&
                  !nonInternalValues.containsKey(e.getKey) // exclude keys already in nonInternalValues
              )
            java.util.stream.Stream.concat(nonInternalValues.entrySet().stream(), originalsFiltered)
          case m: java.util.Map[_, _] =>
            m.asInstanceOf[java.util.Map[String, String]].entrySet().stream()
          case _ => throw new IllegalArgumentException("Unsupported configs type")
        }

        val configEntries = stream
          .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()

Review comment (refactor):
Using `Any` type would disable compiler from checking the input type. Could you please do a bit refactor? for example:
```
      def createResponseResponse(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)
      }

      def createResponseConfig(configs: AbstractConfig,
                               createConfigEntry: (String, Any) => DescribeConfigsResponseData.DescribeConfigsResourceResult): DescribeConfigsResponseData.DescribeConfigsResult = {
        val nonInternalValues = configs.nonInternalValues
        val originalsFiltered = configs.originals.entrySet.stream()
          .filter(e =>
            e.getValue != null &&
              !nonInternalValues.containsKey(e.getKey) // exclude keys already in nonInternalValues
          )
        createResponseResponse(java.util.stream.Stream.concat(nonInternalValues.entrySet().stream(), originalsFiltered), createConfigEntry)
      }

      def createResponseConfig(configs: java.util.Map[String, String],
                               createConfigEntry: (String, Any) => DescribeConfigsResponseData.DescribeConfigsResourceResult): DescribeConfigsResponseData.DescribeConfigsResult = {
        createResponseResponse(configs.entrySet().stream(), createConfigEntry)
      }
```

Code after:
          authHelper.authorize(request.context, DESCRIBE_CONFIGS, TOPIC, resource.resourceName)
        case ConfigResource.Type.GROUP =>
          authHelper.authorize(request.context, DESCRIBE_CONFIGS, GROUP, resource.resourceName)
        case rt => throw new InvalidRequestException(s"Unexpected resource type $rt for resource ${resource.resourceName}")
      }
    }
    val authorizedConfigs = describeConfigs(authorizedResources.toList, describeConfigsRequest.data.includeSynonyms, describeConfigsRequest.data.includeDocumentation)
    val unauthorizedConfigs = unauthorizedResources.map { resource =>
      val error = ConfigResource.Type.forId(resource.resourceType) match {
        case ConfigResource.Type.BROKER | ConfigResource.Type.BROKER_LOGGER | ConfigResource.Type.CLIENT_METRICS => Errors.CLUSTER_AUTHORIZATION_FAILED
        case ConfigResource.Type.TOPIC => Errors.TOPIC_AUTHORIZATION_FAILED
        case ConfigResource.Type.GROUP => Errors.GROUP_AUTHORIZATION_FAILED
        case rt => throw new InvalidRequestException(s"Unexpected resource type $rt for resource ${resource.resourceName}")
      }
      new DescribeConfigsResponseData.DescribeConfigsResult().setErrorCode(error.code)
        .setErrorMessage(error.message)
        .setConfigs(Collections.emptyList[DescribeConfigsResponseData.DescribeConfigsResourceResult])
        .setResourceName(resource.resourceName)
        .setResourceType(resource.resourceType)
    }
    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}")