CODEWITHSUNDEEP
apache-kafkakafka-consumer-api
makados
makados

Reputation: 149

How to find out the latest offset of a Kafka topic to know when my reader is up-to-date with topic?

I have a server that needs to keep an in-memory cache of all users. So assuming that a list won't be big - couple hundred thousands items, I'd like to use a Kafka topic with keyed messages where key is a userId to keep the current state of that list and the admin application will send new user object to that topic when something changed. So when the server starts it simply needs to read everything from that topic from the beginning and populate it's cache.

The population phase takes about 20-30 seconds depending on a connection to Kafka so the server needs not become online until it reads everything from the topic to have an up-to-date cache (all the messages in the topic at the moment of start is considered up-to-date). But I don't see how to determine if I read everything from Kafka stream to notify other services that cache is populated and the server can start server requests. I've read about high watermark but don't see it exposed in Java consumer API.

So how to find out the latest offset of a Kafka topic to know when my reader is up-to-date?

Upvotes: 3

Views: 5631

Answers (1)

mrnakumar
mrnakumar

Reputation: 657

Assuming you are using High level consumer.

High watermark is not available in High level consumer. 

**As you mentioned: all the messages in the topic at the moment of start is considered up-to-date**

when your application starts, you can do the following using SimpleConsumer Api :-

  1. Find the number of partitions in topic by issuing a TopicMetadataRequest to any broker in the kafka cluster.

  2. Create partition to latestOffset map, where key is partition and value is latestOffset available in that partition.

    Map<Integer,Integer> offsetMap = new HashMap<>()

  3. For each partition p in Topic:

    A. Find the leader of partition p

    B. Send an OffsetRequest to the leader

    C. Get the latestOffset from the OffsetResponse

    D. Add an entry to offsetMap where key is partition p and offset is latestOffset.

  4. Start reading messages from kafka using High level consumer:

    A. For each message you get from KafkaStream:

      AA. Get the partition && offset of the message
  BB. if( offsetMap.get(partition)<=offset) stop Reading from this steam

Hope this helps.

Upvotes: 5

Related Questions