How do streams handle consumer confirmations?

[laststem]

What happens if I don't receive the message the broker sent?

1. Stream consumer connected broker 1 (subscribe)
2. broker 1 sent message to consumer (offset 0), consumer got message and handled
3. broker 1 sent message to consumer (offset 1), consumer got message and handled
4. broker 1 sent message to consumer (offset 2), consumer didnt receive message (package loss)
5. broker 1 sent message to consumer (offset 3), consumer got message but unexpected offset

So, if there is no mechanism recover message lost, what can consumer do?

[michaelklishin]

@laststem we need an executable way to reproduce. We will not guess what your code does, what client library you use, and so on.

We do not guess in this community.

[michaelklishin]

If this is a general question about the stream protocol and how it handles confirms on the consumer side, then after step 4 the connection will very likely be be considered closed (the general problem is described in the Heartbeats guide).

The consumer can recover (open a new connection) and begin consuming again. There is a brief mention of this in Stream tutorial 2:

RabbitMQ Streams provide server-side offset tracking to store the progress of
a given consumer in a stream. If the consumer were to stop for
any reason (crash, upgrade, etc), it would be able to re-attach
where it stopped previously to avoid processing the same messages.

In other words, like with any other protocol, consumers sometimes will have to be ready to recover
from different failure scenarios, and re-register. Streams can be consumed repeatedly, so if anything, they depend less on the practical efficiency of the heartbeat mechanism than other protocols,
and the "re-queueing" process for them is simpler in part because of the "repeatable reads" characteristic.

[laststem]

I agree with we can recover offset by open a new connection with server side offset.

But doing so doesn't seem to be very usable.

If step5 occurs, it would be better to have an API that seeks the offset than reconnect.

var expectedNextOffset = 2 // Step 4, expected receive offset 2
fun listen(message: Message, ctx: MessageHandler.Context) {
  val offset = ctx.offset() // Step 5, didnt received offset 2 cause packet loss, offset 3 received
  if (offset != expectedNextOffset) {
    ctx.seek(expectedNextOffset)
    return
  }

  expectedNextOffset += 1
  handle(message)
}

[michaelklishin]

You won't receive "offset 3" before "offset 2" due to how TCP packet retransmission and the heartbeat mechanisms in messaging protocols work. For all other failures, the consumer must be developed with recovery in mind.

[acogoluegnes]

Note message offset may not be contiguous, because a stream can contain other entries than messages and those entries will have their own offsets. Don't base your logic on offset contiguity. You can use your own sequence or check offsets always move forward (but they may actually go backward in case of duplicate messages, which are unlikely but can happen).

[laststem]

Thanks for reply.

I was curious because I had an experience where I couldn't receive a response due to an error that the packet did not arrive when communicating with a TCP-based system like MySQL.

In theory, I agree that the order is maintained through retransmission.

[laststem]

In docs, It is described that the network may fail then message lost.

Applying the examples 1 to 5 above, if we consider the broker as a publisher, isn't it the same as saying that we can't guarantee that the stream message sent by the broker to the consumer will arrive? (situation 4)

I agree that it is recoverable since the consumer can specify an offset to start from if the connection is re-created.

But I'm wondering how to handle the case where the connection is alive but some messages are lost on the network.

thanks

[kjnilsson]

if the connection is alive this wont happen. See TCP

[michaelklishin]

@laststem we have said that TCP takes care of retransmission and ordering of segments at least twice in this thread, e.g. in #12126 (reply in thread).

RabbitMQ Stream Protocol assumes that reliable ordered delivery of TCP segments is a problem solved a long time ago. Heartbeats then simply introduce a way to detect (potentially, likely) unavailable peers quicker because TCP defaults in most kernels are absurdly high for modern distributed data services.

[kjnilsson]

What happens if I don't receive the message the broker sent?

Why would you not receive it? Is this something you have seen in use / testing? Please provide an executable example.