java_kafkasamples/src/main/java/KafkaConsumerCommon.java

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.serialization.StringDeserializer;

public class KafkaConsumerCommon {
  public static final String STRING_DESERIALIZER = StringDeserializer.class.getName();
  public static final Integer TIMEOUT = 200;
  protected static Boolean started = false;
  protected static Integer pass = 0;

  static void process(ConsumerRecords<String, String> records) {
    if (!started) {
      started = true;
      System.out.printf("Started");
    }
//                for (ConsumerRecord<String, String> cr : records.records(TOPIC)) {
//                    System.out.printf("\t\tKey: %s Value: %s\n", cr.key(), cr.value());
//                }
    for (ConsumerRecord<String, String> cr : records) {
      System.out.printf("Pass: %d/%d Topic: %s Partition: %d, Offset: %d, Key: %s Value: %s\n",
        pass, records.count(),
        cr.key(), cr.partition(), cr.offset(), cr.key(), cr.value());
    }
    pass++;
  }
}