def describe_configs(self, config_resources, include_synonyms=False):
        """Fetch configuration parameters for one or more Kafka resources.

        Arguments:
            config_resources: An list of ConfigResource objects.
                Any keys in ConfigResource.configs dict will be used to filter the
                result. Setting the configs dict to None will get all values. An
                empty dict will get zero values (as per Kafka protocol).

        Keyword Arguments:
            include_synonyms (bool, optional): If True, return synonyms in response. Not
                supported by all versions. Default: False.

        Returns:
            Appropriate version of DescribeConfigsResponse class.
        """

        # Break up requests by type - a broker config request must be sent to the specific broker.
        # All other (currently just topic resources) can be sent to any broker.
        broker_resources = []
        topic_resources = []

        for config_resource in config_resources:
            if config_resource.resource_type == ConfigResourceType.BROKER:
                broker_resources.append(self._convert_describe_config_resource_request(config_resource))
            else:
                topic_resources.append(self._convert_describe_config_resource_request(config_resource))

        futures = []
        version = self._client.api_version(DescribeConfigsRequest, max_version=2)
        if version == 0:
            if include_synonyms:
                raise IncompatibleBrokerVersion(
                    "include_synonyms requires DescribeConfigsRequest >= v1, which is not supported by Kafka {}."
                        .format(self.config['api_version']))

            if len(broker_resources) > 0:
                for broker_resource in broker_resources:
                    try:
                        broker_id = int(broker_resource[1])
                    except ValueError:
                        raise ValueError("Broker resource names must be an integer or a string represented integer")

                    futures.append(self._send_request_to_node(
                        broker_id,
                        DescribeConfigsRequest[version](resources=[broker_resource])
                    ))

            if len(topic_resources) > 0:
                futures.append(self._send_request_to_node(
                    self._client.least_loaded_node(),
                    DescribeConfigsRequest[version](resources=topic_resources)
                ))

        elif version <= 2:
            if len(broker_resources) > 0:
                for broker_resource in broker_resources:
                    try:
                        broker_id = int(broker_resource[1])
                    except ValueError:
                        raise ValueError("Broker resource names must be an integer or a string represented integer")

                    futures.append(self._send_request_to_node(
                        broker_id,
                        DescribeConfigsRequest[version](
                            resources=[broker_resource],
                            include_synonyms=include_synonyms)
                    ))

            if len(topic_resources) > 0:
                futures.append(self._send_request_to_node(
                    self._client.least_loaded_node(),
                    DescribeConfigsRequest[version](resources=topic_resources, include_synonyms=include_synonyms)
                ))
        else:
            raise NotImplementedError(
                "Support for DescribeConfigs v{} has not yet been added to KafkaAdminClient.".format(version))

        self._wait_for_futures(futures)
        return [f.value for f in futures]
Fetch configuration parameters for one or more Kafka resources.

        Arguments:
            config_resources: An list of ConfigResource objects.
                Any keys in ConfigResource.configs dict will be used to filter the
                result. Setting the configs dict to None will get all values. An
                empty dict will get zero values (as per Kafka protocol).

        Keyword Arguments:
            include_synonyms (bool, optional): If True, return synonyms in response. Not
                supported by all versions. Default: False.

        Returns:
            Appropriate version of DescribeConfigsResponse class.
        
kafka/admin/client.py