CODEWITHSUNDEEP
javaapache-kafka
Zack Macomber
Zack Macomber

Reputation: 6905

How to reliably get all kafka topic messages via KafkaConsumer

The getMessages() method below sometimes gets all messages for a kafka topic. This code is executed in a web application on page load. Sometimes no messages come back and sometimes all messages come back.

Is there a way to set properties and/or change the code so that all messages come back every time?

public List<String> getMessages() {
    List<String> messages = new ArrayList<>();
    try {
        ConnectionKafka connection = ConstantsHome.connectionManager.getConnectionDef(getGuid(), ConnectionKafka.class);
        Properties props = new Properties();
        props.put("bootstrap.servers", connection.getProps().get("bootstrapServers"));
        props.put("group.id", getName());
        props.put("auto.offset.reset", "earliest");
        props.put("enable.auto.commit", "true");
        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);
        consumer.subscribe(Collections.singleton(getName()));
        consumer.poll(0);
        consumer.seekToBeginning(consumer.assignment());
        ConsumerRecords<String, String> records = consumer.poll(0);
        for (ConsumerRecord<String, String> record : records) {
            messages.add(
                String.format("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value())
            );
        }
        consumer.close(0, TimeUnit.MILLISECONDS);
    } catch (Exception e) {
        Utils.writeToLog(e, getClass().getName(), "", IErrorManager.ERROR);
    }
    Collections.sort(messages, new Comparator<String>() {
        @Override
        public int compare(String o1, String o2) {
            return Integer.valueOf(o1.substring("offset = ".length(), o1.indexOf(","))) -
            Integer.valueOf(o2.substring("offset = ".length(), o2.indexOf(",")));
        }
    });
    return messages;
}

Upvotes: 0

Views: 2264

Answers (1)

hlagos
hlagos

Reputation: 7947

If your expectation is get all the messages for each call, you should be setting the following propertly

enable.auto.commit = false

The other option is create a dynamic group ID for each iteration, I would avoid this option considering that the groups metadata is stored in the kafka side.

Upvotes: 1

Related Questions