Search code examples
streamrabbitmqoffset

How to make use of x-stream-offset when consuming a RabbitMQ Stream to keep track of what has been processed?

The official documentation states that one can use the x-stream-offset consumer argument to specify where to start reading the Stream from. But it doesn't say where to get these values from.

For example, what does timestamp mean here? The point in time when the chunk was inserted into the stream? If one is reading messages from the Stream, is there a way to get this timestamp to store it so that when the process is restarted it can continue from where it left off?

Same with the numerical offset value... how does a consumer know the current offset when reading messages from the stream?

Solution

  • The offset for each message can be retrieved from its headers.

    Usually:

    message.properties.headers["x-stream-offset"]