(self, api_version)
return kafka_admin_client
Return a KafkaAdminClient.
Return a KafkaAdminClient.
Return a KafkaAdminClient .
def _create_kafka_admin_client(self, api_version):
        """Return a KafkaAdminClient."""
        # TODO accept None (which inherits kafka-python default of localhost:9092)
        kafka_admin_client = self.create_kafka_admin_client()
        self.log.debug("KafkaAdminClient api_version: %s", kafka_admin_client.config['api_version'])
        # Force initial population of the local cluster metadata cache
        kafka_admin_client._client.poll(future=kafka_admin_client._client.cluster.request_update())
        if kafka_admin_client._client.cluster.topics(exclude_internal_topics=False) is None:
            raise RuntimeError("Local cluster metadata cache did not populate.")
        return kafka_admin_client
def _create_kafka_admin_client ( self , api_version ) : # TODO accept None (which inherits kafka-python default of localhost:9092) kafka_admin_client = self . create_kafka_admin_client ( ) self . log . debug ( "KafkaAdminClient api_version: %s" , kafka_admin_client . config [ 'api_version' ] ) # Force initial population of the local cluster metadata cache kafka_admin_client . _client . poll ( future = kafka_admin_client . _client . cluster . request_update ( ) ) if kafka_admin_client . _client . cluster . topics ( exclude_internal_topics = False ) is None : raise RuntimeError ( "Local cluster metadata cache did not populate." ) return kafka_admin_client