Kafka Consumer API Usage Sample

Function Description

The following code snippet belongs to the run method in the com.huawei.bigdata.kafka.example.Consumer class. It is used to consume topic messages that are subscribed to.

Sample Code

DoWork method logic of the consumer thread. This method is the rewrite of the run method.

    /**
     * Message processing function for subscribing to topics
     */
    public void doWork()
    {
        // Subscribe.
        consumer.subscribe(Collections.singletonList(this.topic));
        // Message consumption request
        ConsumerRecords<Integer, String> records = consumer.poll(waitTime);
        // Message processing
        for (ConsumerRecord<Integer, String> record : records)
        {
            LOG.info("[NewConsumerExample], Received message: (" + record.key() + ", " + record.value()
                + ") at offset " + record.offset());
        }
    }