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

Code before:
        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: JMap[String, _ <: Object],
                               createConfigEntry: (String, Object) => DescribeConfigsResponseData.DescribeConfigsResourceResult): DescribeConfigsResponseData.DescribeConfigsResult = {
        val keySet: HashSet[String] =
          if (resource.configurationKeys == null || resource.configurationKeys.isEmpty) null
          else new HashSet[String](resource.configurationKeys)

        val configEntries: JList[DescribeConfigsResponseData.DescribeConfigsResourceResult] = {
          val baseStream = configs.entrySet().stream()
          val filtered = if (keySet == null) baseStream
          else baseStream.filter(entry => keySet.contains(entry.getKey))
          filtered
            .map[DescribeConfigsResponseData.DescribeConfigsResourceResult](entry =>
              createConfigEntry(entry.getKey, entry.getValue)
            )
            .collect(Collectors.toList())
        }

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

      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(allConfigs(logConfig), createTopicConfigEntry(logConfig, topicProps, includeSynonyms, includeDocumentation))
            } else {
              new DescribeConfigsResponseData.DescribeConfigsResult().setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code)
                .setConfigs(Collections.emptyList[DescribeConfigsResponseData.DescribeConfigsResourceResult])
            }

Review comment (suggestion):
I think we can avoid using null checks to distinguish between different filters.

Code after:
            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)(_, _))
            }

          case ConfigResource.Type.GROUP =>
            val group = resource.resourceName
            if (group == null || group.isEmpty) {
              throw new InvalidRequestException("Group name must not be empty")
            } else {
              val groupProps = configRepository.groupConfig(group)
              val groupConfig = GroupConfig.fromProps(config.groupCoordinatorConfig.extractGroupConfigMap(config.shareGroupConfig), groupProps)
              createResponseConfig(resource, groupConfig, createGroupConfigEntry(groupConfig, groupProps, includeSynonyms, includeDocumentation)(_, _))
            }

          case resourceType => throw new InvalidRequestException(s"Unsupported resource type: $resourceType")