Consuming Messages

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "billing");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

props.put("enable.auto.commit", "false");
props.put("auto.offset.reset", "earliest");

props.put("bootstrap.servers", "kafka1:9092,kafka2:9092");

props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
props.put("schema.registry.url", "http://localhost:8081");
props.put("specific.avro.reader", "true");

props.put("group.id", "billing");
props.put("group.instance.id", "billing-pod-1");

consumer.subscribe(List.of("orders", "refunds"));
// Or pattern-based:
consumer.subscribe(Pattern.compile("metrics-.*"));

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
    for (ConsumerRecord<String, String> r : records) process(r);
}

for (ConsumerRecord<String, String> r : records) {
    log.info("p={} o={} k={} v={}", r.partition(), r.offset(), r.key(), r.value());
}

TopicPartition tp = new TopicPartition("orders", 0);
consumer.assign(List.of(tp));

consumer.seek(new TopicPartition("orders", 0), 1500L);
consumer.seekToBeginning(consumer.assignment());

Runtime.getRuntime().addShutdownHook(new Thread(consumer::wakeup));
try { /* poll loop */ }
catch (WakeupException e) { /* expected */ }
finally { consumer.close(); }