CODEWITHSUNDEEP
javaapache-kafkaspring-kafka
BanzaiTokyo
BanzaiTokyo

Reputation: 1404

Is there a way to get the last message from Kafka topic?

I have a Kafka topic with multiple partitions and I wonder if there is a way in Java to fetch the last message for the topic. I don't care for the partitions I just want to get the latest message.

I have tried @KafkaListener but it fetches the message only when the topic is updated. If there is nothing published after the application is opened nothing is returned.

Maybe the listener is not the right approach to the problem at all?

Upvotes: 5

Views: 15433

Answers (2)

JavaTechnical
JavaTechnical

Reputation: 9347

This following snippet worked for me. You may try this. Explanation in the comments. 

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Collections.singletonList(topic));

        consumer.poll(Duration.ofSeconds(10));

        consumer.assignment().forEach(System.out::println);

        AtomicLong maxTimestamp = new AtomicLong();
        AtomicReference<ConsumerRecord<String, String>> latestRecord = new AtomicReference<>();

        // get the last offsets for each partition
        consumer.endOffsets(consumer.assignment()).forEach((topicPartition, offset) -> {
            System.out.println("offset: "+offset);

            // seek to the last offset of each partition
            consumer.seek(topicPartition, (offset==0) ? offset:offset - 1);

            // poll to get the last record in each partition
            consumer.poll(Duration.ofSeconds(10)).forEach(record -> {

                // the latest record in the 'topic' is the one with the highest timestamp
                if (record.timestamp() > maxTimestamp.get()) {
                    maxTimestamp.set(record.timestamp());
                    latestRecord.set(record);
                }
            });
        });
        System.out.println(latestRecord.get());

Upvotes: 6

Prashant Pandey
Prashant Pandey

Reputation: 4642

You'll have to consume the latest message from each partition and then do a comparison on the client side (using the timestamp on the message, if it contains it). The reason for this is that Kafka does not guarantee inter-partition ordering. Inside a partition, you can be sure that the message with the largest offset is the latest message pushed to it.

Upvotes: 1

Related Questions