KafkaAvroDemo.java 文件源码

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