def _get_cluster_metadata(self, topics=None, auto_topic_creation=False):
        """
        topics == None means "get all topics"
        """
        version = self._client.api_version(MetadataRequest, max_version=5)
        if version <= 3:
            if auto_topic_creation:
                raise IncompatibleBrokerVersion(
                    "auto_topic_creation requires MetadataRequest >= v4, which"
                    " is not supported by Kafka {}"
                    .format(self.config['api_version']))

            request = MetadataRequest[version](topics=topics)
        elif version <= 5:
            request = MetadataRequest[version](
                topics=topics,
                allow_auto_topic_creation=auto_topic_creation
            )

        future = self._send_request_to_node(
            self._client.least_loaded_node(),
            request
        )
        self._wait_for_futures([future])
        return future.value

        topics == None means "get all topics"
        
kafka/admin/client.py