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 records) { if (!started) { started = true; System.out.printf("Started"); } // for (ConsumerRecord cr : records.records(TOPIC)) { // System.out.printf("\t\tKey: %s Value: %s\n", cr.key(), cr.value()); // } for (ConsumerRecord 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++; } }