Consume messages from kafka

           Consume messages from kafka using the Kazoo SetPartitioner to
           allow multiple consumer processes to negotiate access to the kafka
           partitions
168069
def _partition(self):
        

        # KazooClient and SetPartitioner objects need to be instantiated after
        # the consumer process has forked.  Instantiating prior to forking
        # gives the appearance that things are working but after forking the
        # connection to zookeeper is lost and no state changes are visible

        if not self._kazoo_client:
            self._kazoo_client = KazooClient(hosts=self._zookeeper_url)
            self._kazoo_client.start()

            state_change_event = threading.Event()

            self._set_partitioner = (
                SetPartitioner(self._kazoo_client,
                               path=self._zookeeper_path,
                               set=self._consumer.fetch_offsets.keys(),
                               state_change_event=state_change_event,
                               identifier=str(datetime.datetime.now())))

        try:
            while 1:
                if self._set_partitioner.failed:
                    raise Exception("Failed to acquire partition")

                elif self._set_partitioner.release:
                    log.info("Releasing locks on partition set {} "
                             "for topic {}".format(self._partitions,
                                                   self._kafka_topic))
                    self._set_partitioner.release_set()

                    self._partitions = []

                elif self._set_partitioner.acquired:
                    if not self._partitions:
                        self._partitions = [p for p in self._set_partitioner]

                        if not self._partitions:
                            log.info("Not assigned any partitions on topic {},"
                                     " waiting for a Partitioner state change"
                                     .format(self._kafka_topic))
                            state_change_event.wait()
                            state_change_event.clear()
                            continue

                        log.info("Acquired locks on partition set {} "
                                 "for topic {}".format(self._partitions, self._kafka_topic))

                        # Reconstruct the kafka consumer object because the
                        # consumer has no API that allows the set of partitons
                        # to be updated outside of construction.
                        self._consumer.stop()
                        self._consumer = self._create_kafka_consumer(self._partitions)
                        return

                elif self._set_partitioner.allocating:
                    log.info("Waiting to acquire locks on partition set")
                    self._set_partitioner.wait_for_acquire()

        except Exception:
            log.exception('KafkaConsumer encountered fatal exception '
                          'processing messages.')
            raise
624897
public KafkaMessage sendMessage(ProducerType type, KafkaMessage message) {
        String key = message.key();
        ProducerRecord<String, byte[]> record = StringUtils.isEmpty(key)
                ? new ProducerRecord<>(message.topic(), message.content())
                : new ProducerRecord<>(message.topic(), key, message.content());
        KafkaProducer<String, byte[]> producer = getJavaProducer(type);
        try {
            RecordMetadata metadata = producer.send(record).get();
            KafkaMessage kafkaMessage = new KafkaMessage(message);
            kafkaMessage.partition(metadata.partition());
            kafkaMessage.offset(metadata.offset());
            return kafkaMessage;
        } catch (InterruptedException | ExecutionException e) {
            throw new KafkaException(e);
        }
    }
182313
def get_kafka_groups(cls, cluster_config):
        ''''''
        kafka_group_reader = KafkaGroupReader(cluster_config)
        return list(kafka_group_reader.read_groups().keys())
1236792
def consume(self, queue, consumer, consumer_tag='', no_local=False,
                no_ack=True, exclusive=False, nowait=True, ticket=None,
                cb=None):
        '''
        
        '''
        nowait = nowait and self.allow_nowait() and not cb

        if nowait and consumer_tag == '':
            consumer_tag = self._generate_consumer_tag()

        args = Writer()
        args.write_short(ticket or self.default_ticket).\
            write_shortstr(queue).\
            write_shortstr(consumer_tag).\
            write_bits(no_local, no_ack, exclusive, nowait).\
            write_table({})  # unused according to spec
        self.send_frame(MethodFrame(self.channel_id, 60, 20, args))

        if not nowait:
            self._pending_consumers.append((consumer, cb))
            self.channel.add_synchronous_cb(self._recv_consume_ok)
        else:
            self._consumer_cb[consumer_tag] = consumer
786994
public KafkaMessage consumeMessage(String consumerGroupId, String topic) {
        return consumeMessage(consumerGroupId, true, topic);
    }
226597
def subscribe(*topics)
      # Create topic partition list with topics and no partition set
      tpl = TopicPartitionList.new_native_tpl(topics.length)

      topics.each do |topic|
        Rdkafka::Bindings.rd_kafka_topic_partition_list_add(
          tpl,
          topic,
          -1
        )
      end
      # Subscribe to topic partition list and check this was successful
      response = Rdkafka::Bindings.rd_kafka_subscribe(@native_kafka, tpl)
      if response != 0
        raise Rdkafka::RdkafkaError.new(response, "Error subscribing to '#{topics.join(', ')}'")
      end
    end
1307813
def consume_network_packet_messages_from_redis():
    
    # end of recv_message
    # Initialize KombuSubscriber
    sub = KombuSubscriber(
        name,
        FORWARD_BROKER_URL,
        FORWARD_SSL_OPTIONS)

    # Now consume:
    seconds_to_consume = 10.0
    heartbeat = 60
    serializer = "application/json"
    queue = FORWARD_QUEUE

    sub.consume(
        callback=recv_msg,
        queue=queue,
        exchange=None,
        routing_key=None,
        serializer=serializer,
        heartbeat=heartbeat,
        time_to_wait=seconds_to_consume)

    log.info("end - {}".format(name))
322200
public boolean seek(String consumerGroupId, KafkaTopicPartitionOffset tpo) {
        KafkaMsgConsumer consumer = getKafkaConsumer(consumerGroupId, false);
        return consumer.seek(tpo);
    }
63220
def connect_to_kafka(self, bootstrap_servers='127.0.0.1:9092',
                         auto_offset_reset='latest',
                         client_id='Robot',
                         **kwargs
                         ):
        

        self.connect_consumer(
            bootstrap_servers=bootstrap_servers,
            auto_offset_reset=auto_offset_reset,
            client_id=client_id,
            **kwargs
        )
        self.connect_producer(bootstrap_servers=bootstrap_servers, client_id=client_id)
416473
public static void setOffsetInZooKeeper(CuratorFramework curatorClient, String groupId, String topic, int partition, long offset) throws Exception {
		ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(groupId, topic);
		String path = topicDirs.consumerOffsetDir() + "/" + partition;
		curatorClient.newNamespaceAwareEnsurePath(path).ensure(curatorClient.getZookeeperClient());
		byte[] data = Long.toString(offset).getBytes(ConfigConstants.DEFAULT_CHARSET);
		curatorClient.setData().forPath(path, data);
	}
1179175
private KafkaConsumer<String, byte[]> _getConsumer(String topic, boolean autoCommitOffsets) {
        KafkaConsumer<String, byte[]> consumer = topicConsumers.get(topic);
        if (consumer == null) {
            consumer = KafkaHelper.createKafkaConsumer(bootstrapServers, consumerGroupId,
                    consumeFromBeginning, autoCommitOffsets, consumerProperties);
            KafkaConsumer<String, byte[]> existingConsumer = topicConsumers.putIfAbsent(topic,
                    consumer);
            if (existingConsumer != null) {
                consumer.close();
                consumer = existingConsumer;
            } else {
                _checkAndSubscribe(consumer, topic);
            }
        }
        return consumer;
    }
639338
public Kafka sinkPartitionerCustom(Class<? extends FlinkKafkaPartitioner> partitionerClass) {
		sinkPartitionerType = CONNECTOR_SINK_PARTITIONER_VALUE_CUSTOM;
		sinkPartitionerClass = Preconditions.checkNotNull(partitionerClass);
		return this;
	}
1073279
def _create_producer(self):
        """"""
        if not self.closed:
            try:
                self.logger.debug("Creating new kafka producer using brokers: " +
                                   str(self.settings['KAFKA_HOSTS']))

                return KafkaProducer(bootstrap_servers=self.settings['KAFKA_HOSTS'],
                                     value_serializer=lambda v: json.dumps(v).encode('utf-8'),
                                     retries=3,
                                     linger_ms=self.settings['KAFKA_PRODUCER_BATCH_LINGER_MS'],
                                     buffer_memory=self.settings['KAFKA_PRODUCER_BUFFER_BYTES'])
            except KeyError as e:
                self.logger.error('Missing setting named ' + str(e),
                                   {'ex': traceback.format_exc()})
            except:
                self.logger.error("Couldn't initialize kafka producer.",
                                   {'ex': traceback.format_exc()})
                raise
482728
function createKafkaConsumerAsync(kafkaConfig) {
    let topicConfig = {};
    if ('default_topic_config' in kafkaConfig) {
        topicConfig = kafkaConfig.default_topic_config;
        delete kafkaConfig.default_topic_config;
    }

    const consumer = P.promisifyAll(
        new kafka.KafkaConsumer(kafkaConfig, topicConfig)
    );

    return consumer.connectAsync(undefined)
    .thenReturn(consumer);
}
1182000
def _setup_kafka(self):
        
        # close older connections
        if self.consumer is not None:
            self.logger.debug("Closing existing kafka consumer")
            self.consumer.close()
            self.consumer = None
        if self.producer is not None:
            self.logger.debug("Closing existing kafka producer")
            self.producer.flush()
            self.producer.close(timeout=10)
            self.producer = None

        # create new connections
        self._consumer_thread = None
        self.logger.debug("Creating kafka connections")
        self.consumer = self._create_consumer()
        if not self.closed:
            self.logger.debug("Kafka Conumer created")
        self.producer = self._create_producer()
        if not self.closed:
            self.logger.debug("Kafka Producer created")

        if not self.closed:
            self.kafka_connected = True
            self.logger.info("Connected successfully to Kafka")
            self._spawn_kafka_consumer_thread()
513605
def start(self):
        '''
        
        '''
        log.debug('Creating the consumer using the bootstrap servers: %s and the group ID: %s',
                  self.bootstrap_servers,
                  self.group_id)
        try:
            self.consumer = kafka.KafkaConsumer(bootstrap_servers=self.bootstrap_servers,
                                                group_id=self.group_id)
        except kafka.errors.NoBrokersAvailable as err:
            log.error(err, exc_info=True)
            raise ListenerException(err)
        log.debug('Subscribing to the %s topic', self.topic)
        self.consumer.subscribe(topics=[self.topic])
1259995
def _send_to_kafka(self, master):
        '''
        
        '''
        appid_topic = "{prefix}.outbound_{appid}".format(
                                                    prefix=self.topic_prefix,
                                                    appid=master['appid'])
        firehose_topic = "{prefix}.outbound_firehose".format(
                                                    prefix=self.topic_prefix)
        try:
            # dont want logger in outbound kafka message
            if self.use_appid_topics:
                f1 = self.producer.send(appid_topic, master)
                f1.add_callback(self._kafka_success)
                f1.add_errback(self._kafka_failure)
            f2 = self.producer.send(firehose_topic, master)
            f2.add_callback(self._kafka_success)
            f2.add_errback(self._kafka_failure)

            return True
        except Exception as ex:
            message = "An exception '{0}' occured while sending a message " \
                "to kafka. Arguments:\n{1!r}" \
                .format(type(ex).__name__, ex.args)
            self.logger.error(message)

        return False