CompatibleConsumer.java

CompatibleConsumer.java

try {
    consumer.subscribe(Collections.singletonList( topicName ) );

    while(true) {
        Duration duration = Duration.ofMillis(1000);
        ConsumerRecords<Long, String> consumerRecords = consumer.poll(duration);
        consumerRecords.forEach(record -> {
            System.out.println("Record Key " + record.key());
            System.out.println("Record value " + record.value());
            System.out.println("Record partition " + record.partition());
            System.out.println("Record offset " + record.offset());
        });
        // commits the offset of record to broker.
        consumer.commitAsync();
    }
}
catch(WakeupException e) {
    // do nothing, shutting down...
}
finally {
    System.out.println("closing consumer");
    consumer.close();
}