KafkaStringDemo.java 文件源码

public void run() {
    ConsumerIterator<byte[], byte[]> it = kafkaStream.iterator();
    while (it.hasNext()) {
        MessageAndMetadata<byte[], byte[]> messageAndMetadata = it.next();
        String key = new String(  messageAndMetadata.key() );
        String message = new String(  messageAndMetadata.message() );
        String summary = 
                "Thread " + threadNumber + 
                ", topic=" + messageAndMetadata.topic() + 
                ", partition=" + messageAndMetadata.partition() + 
                ", key=" + key + 
                ", message=" + message + 
                ", offset=" + messageAndMetadata.offset() + 
                ", timestamp=" + messageAndMetadata.timestamp() + 
                ", timestampType=" + messageAndMetadata.timestampType();
        logger.info(">>> Consumed: " + summary);
    }
    logger.info(">>> Shutting down Thread: " + threadNumber);
}