(self, timeout=MIN_TIMEOUT, auto_offset_reset='smallest')
return self._consumer
Allow re-initing the consumer if necessary
Allow re-initing the consumer if necessary
Allow re - initing the consumer if necessary
def _init_consumer(self, timeout=MIN_TIMEOUT, auto_offset_reset='smallest'):
        """Allow re-initing the consumer if necessary
        """
        config = {
            'client_id': self._client_id,
            'bootstrap_servers': settings.KAFKA_BROKERS,
            'consumer_timeout_ms': timeout,
            'auto_offset_reset': auto_offset_reset,
            'enable_auto_commit': False,
            'api_version': settings.KAFKA_API_VERSION,
        }
        self._consumer = KafkaConsumer(**config)

        topic_partitions = []
        for topic in self.topics:
            for partition in self._consumer.partitions_for_topic(topic):
                topic_partitions.append(TopicPartition(topic, partition))

        self._consumer.assign(self._filter_partitions(topic_partitions))
        return self._consumer
def _init_consumer ( self , timeout = MIN_TIMEOUT , auto_offset_reset = 'smallest' ) : config = { 'client_id' : self . _client_id , 'bootstrap_servers' : settings . KAFKA_BROKERS , 'consumer_timeout_ms' : timeout , 'auto_offset_reset' : auto_offset_reset , 'enable_auto_commit' : False , 'api_version' : settings . KAFKA_API_VERSION , } self . _consumer = KafkaConsumer ( * * config ) topic_partitions = [ ] for topic in self . topics : for partition in self . _consumer . partitions_for_topic ( topic ) : topic_partitions . append ( TopicPartition ( topic , partition ) ) self . _consumer . assign ( self . _filter_partitions ( topic_partitions ) ) return self . _consumer