Stream Offsets and Policy Max Age / Max Length

[thaoula]

Hi All,

We have recently started using Streams via AMQP. We are running on NodeJS cannot use the stream protocol.

• We have enabled a policy on the stream to prune it using Max Age and Max Length.
• We are also persisting the x-stream-offset when we receive a new message.

The x-stream-offset appears to be integer similar to array index which is great if the steam is immutable and is never purged.

Our issue that we are unable to resume on restart if the stream is purged due to Max Age or Max Length. The offsets are greater than the messages in the stream.

My questions are -

1. Is there a way to detect the scenario where the start offset is great than the stream x-stream-offset so we can fall back to next or first.
2. 1. If not, is there a way to obtain a stream's offset in code so that we can decided whether to start at the offset or use next or first?

Regards,
Tarek

[kjnilsson]

"The offsets are greater than the messages in the stream." I am not sure what you mean by this? Stream retention will always leave at least one segment worth of data - if you try to attach at a stream offset that is smaller than the first offset available in the stream it will fall back to the first offset already and you don't need to do anything else. If you have further issues please provide reproduction steps. Cheers Karl

…

On Mon, 17 Jul 2023 at 11:22, Tarek Haoula ***@***.***> wrote: Hi All, We have recently started using Streams via AMQP. We are running on NodeJS cannot use the stream protocol. - We have enabled a policy on the stream to prune it using Max Age and Max Length. - We are also persisting the x-stream-offset when we receive a new message. The x-stream-offset appears to be integer similar to array index which is great if the steam is immutable and is never purged. Our issue that we are unable to resume on restart if the stream is purged due to Max Age or Max Length. The offsets are greater than the messages in the stream. My questions are - 1. Is there a way to detect the scenario where the start offset is great than the stream x-stream-offset so we can fall back to next or first. 2. 1. If not, is there a way to obtain a stream's offset in code so that we can decided whether to start at the offset or use next or first? Regards, Tarek — Reply to this email directly, view it on GitHub <#8886>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAJAHFCUJKV4BPRMJG5WXP3XQUG6BANCNFSM6AAAAAA2MXGJMY> . You are receiving this because you are subscribed to this thread.Message ID: ***@***.***>

-- *Karl Nilsson*

[thaoula]

Hi Karl,

Thanks for your reply.

I have attached a screenshot that I hope will make a little more sense.

Here we have a stream with only 6 items.

We also have a single consumer which has requested x-stream-offset of 101 (artificially increased here as i don't know how to force a stream to trim its messages to simulate this situation)

FYI - Our consumers, retrieve the last offset from the database and set it as an argument x-stream-offset when connecting to the stream.

The consumer will not receive a message until the stream reaches that offset and then it continues as normal.

I hope that is clearer.

Regards,
Tarek

[kjnilsson]

The behaviour you describe is correct! ofc it cannot deliver you offsets it doesn't have. when it truncates the data it leaves the newest data intact so the latest offset is always retained. I don't see what the actual problem is.

…

On Mon, 17 Jul 2023 at 11:52, Tarek Haoula ***@***.***> wrote: Hi Karl, I have attached a screenshot that I hope will make a little more sense. Here we have a stream with only 6 items. We also have a single consumer which has requested x-stream-offset of 101 (artificially increased here as i don't know how to force a stream to trim its messages to simulate this situation) Our consumers, retrieve the last offset from the database and set it as an argument x-stream-offset when connecting to the stream. The consumer will not receive a message until the stream reaches that offset. [image: image] <https://user-images.githubusercontent.com/349133/253916856-34bd7e45-f2f5-496c-8863-6f728e1eda41.png> — Reply to this email directly, view it on GitHub <#8886 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAJAHFAV6SEVI5UTYDNIY5DXQUKPBANCNFSM6AAAAAA2MXGJMY> . You are receiving this because you commented.Message ID: ***@***.*** com>

-- *Karl Nilsson*

[thaoula]

Hi @acogoluegnes,

Thanks for your clarification.

Running the command you mentioned shows the following

So i started to look at our configuration and I think, I tracked this down to misconfiguration - missing hostname on our docker compose. We pulled rabbitmq 3.12.1 today which resulted in a new container hostname and new database. I have added a fixed hostname to the config now.

Many thanks @kjnilsson and @acogoluegnes for your responses.

Regards,
Tarek

[acogoluegnes]

OK, so the stored offset was for a previous "instance" of the stream. That cannot work indeed :-)

[thaoula]

Hi @acogoluegnes,

Yes, pulling the new container generate a new hostname because the docker config did not specify a fixed hostname. This would create a new database and empty stream.

When I run the stream_status command you suggested -

1. How would I tell if the stream has been truncated?
2. Also, when does the truncation actually happen. Is this a scheduled process?

Also, many thanks for your help so far. Your clarifications help track down the issue.

Regards,
Tarek

[acogoluegnes]

If the first offset is not 0, it's very likely the stream has been truncated (see first_offset column from stream_status).

Retention kicks in when the current segment file reaches a given size, is closed, and a new segment file is open. At this point, the broker checks if some of the "old" segment files can be deleted.

[acogoluegnes]

There's also a scheduled job that runs every hour to enforce retention (thanks @kjnilsson for reminding me this).