The behavior of the "last" stream offset

[Laurianti]

Describe the bug

The LAST offset in RabbitMQ Streams is expected to return the first message of the last chunk, according to the official documentation. However, in practice, it consistently returns the last published message, even when multiple chunks are present. This behavior contradicts the documented functionality but aligns with our preferred use case.

https://www.rabbitmq.com/blog/2021/09/13/rabbitmq-streams-offset-tracking#the-different-offset-specifications-in-a-stream
https://www.rabbitmq.com/docs/streams#client-operations

Reproduction steps

1. Create a RabbitMQ stream with a custom configuration (e.g., smaller chunk size, larger payload) to force multiple chunks.
2. Publish a significant number of messages
3. Subscribe to the stream using the LAST offset.
4. Observe that the consumer receives the last message published instead of the first message of the last chunk.

Expected behavior

According to the RabbitMQ Stream Core documentation, the LAST offset should return the first message of the last chunk. However, based on our use case, we would prefer that the LAST offset consistently returns the most recently published message, regardless of chunk boundaries.

Additional context

We are using RabbitMQ Stream Core, not the Stream Plugin. Our tests have been conducted with custom stream configurations to force the creation of multiple chunks by adjusting chunk size and message payload. Despite this, the LAST offset consistently returns the last published message rather than the first message of the last chunk. This behavior aligns with our needs, but it contradicts the documentation, and we would appreciate clarification.

[michaelklishin]

@Laurianti we will not guess what your tests do exactly and how do you determine that the message returned is "the last published" and not "the last message in the chunk", or how do you determine that they are not the same message (I'd say with a lot of tests it will be the case).

Please provide an executable way to reproduce against RabbitMQ 4.0.2.

[Laurianti]

1. The code publishes a message to a RabbitMQ stream, adding a metadata field to track the message.
2. It subscribes to the stream using the LAST offset.
3. The consumer receives the message and verifies if it matches the most recently published one by checking the metadata.
4. After verification, the consumer unsubscribes from the stream.
5. The cycle repeats, publishing and subscribing again.
6. If the received message doesn’t match the expected one, an exception is thrown.
7. This process continues for hundreds of thousands of messages.

If needed, we can provide the code for further clarification.

[michaelklishin]

Our tests have been conducted with custom stream configurations to force the creation of multiple
chunks by adjusting chunk size and message payload

Are we supposed to guess what those "custom stream configurations" involved, when they happened w.r.t. publishing, and so on?

If you are looking for help from the core team for free, please provide all the details we need to reproduce exactly what your doing, or take the source code and do your own investigation. This is open source software after all.

[Laurianti]

adjusting chunk size

We tested both with and without custom configurations, specifically using the x-stream-max-segment-size-bytes parameter (https://www.rabbitmq.com/docs/streams#declaring) to divide the stream into multiple segments. The behavior is the same in both cases: the LAST offset returns the last published message, regardless of whether or not the x-stream-max-segment-size-bytes is set.

[mkuratczyk]

Michael asked for an executable way - please provide it. Otherwise it's not clear what you are doing.

Here's an example showing that I get the whole last chunk (starting from the first message of the last chunk) when consuming over AMQP 0.9.1 and AMQP 1.0:

1. Publish over the stream protocol using stream-perf-test:

> stream-perf-test -y 0 -z 2
...
Summary: published 1952679 msg/s, confirmed 1947733 msg/s, consumed 0 msg/s, latency 95th 0 ms, chunk size 0

2. Consume from last using stream-perf-test - it shows the chunk size of 1740 messages:

> stream-perf-test -x 0 -o last
1, published 0 msg/s, confirmed 0 msg/s, consumed 1726 msg/s, latency median/75th/95th/99th 0/0/0/0 ms, chunk size 1740

3. Consume over AMQP 1.0 using omq - it reports it consumed 1740 messages (so all messages in the last chunk):

> omq amqp -x 0 -T /queues/stream --stream-offset last -z 1s
...
2024/10/09 08:22:44 TOTAL CONSUMED consumed=1740 rate=1708.87/s

4. Consume over AMQP 0.9.1 using perf-test - it doesn't show the total but it does show the throughput and calculates it to be 1729 msg/s in the first second in my case:

> perf-test -x 0 -u stream -p -q 10 --stream-consumer-offset last
...
id: test-082426-610, time 1.006 s, received: 1729 msg/s, min/median/75th/95th/99th consumer latency: 0/0/0/0/0 µs

Clearly, using AMQP 0.9.1 and AMQP 1.0, I get all the messages from the last chunk.

[Laurianti]

This answer is already more concrete, thank you. We will provide an example, we are using .NET client.

[michaelklishin]

@Laurianti I did not ask for more "two sentence descriptions", I have asked for a public repo with some code we can run against RabbitMQ 4.0.2. If you are not willing to provide such a repo, that's fine but don't expect any specific responses. The rules are set by the RabbitMQ core team, please cooperate or take your questions elsewhere.

[Laurianti]

@michaelklishin Thank you for your response, you are really kind. Before we proceed with publishing the code, could you kindly provide a link to the official policies of the RabbitMQ team so that we can ensure we align with and follow all of them?

[michaelklishin]

@Laurianti follow your common sense.

We need an executable way to run what you run, and know exactly how and when you change segment size. Our team is small, we need a way to efficiently reproduce what you see.

Guessing is a very time consuming (read: expensive) way to troubleshoot distributed infrastructure software. You are asking a team of OSS maintainers for free help.