Unexpected consumer lag metric despite full message consumption

[hiimjako]

Hello,

I'm observing a persistent consumer offset lag in the RabbitMQ management UI for a stream, even though my consumer application has successfully processed all messages, verified through application log. I'm trying to determine the cause.

Consumer Configuration & Context

Stream client configuration:

• Consumer Type: High-availability (HA) stream consumer.
• No Offset Commits: We are intentionally not committing offsets. The consumer is designed to read the entire stream from the beginning upon every restart.
• Large Message Size: approximately 500Kb to 1MB each (raw file data sent as bytes). This issue does not seem to occur in streams with smaller messages, maybe it is related to this?
• RabbitMQ cluster of 3 nodes
• Environment: This behavior is often observed in our deployed Kubernetes clusters, and sometimes, in local development environments running the same code, but not that often.

As reference, the consumer code used in the service:

	messageHandler := func(_ stream.ConsumerContext, message *amqp.Message) {
		fmt.Println("message consumed!")
		// message processing
	}

	consumer, err := ha.NewReliableConsumer(
		h.streamClient.Environment,
		"topology",
		stream.NewConsumerOptions().
			SetConsumerName(h.env.ServiceID).
			SetClientProvidedName(h.env.ServiceID).
			SetOffset(stream.OffsetSpecification{}.First()).
			SetInitialCredits(10).
			SetCRCCheck(false),
		messageHandler)

Questions

Is this expected behavior? If not, what could be the cause? Could be caused by the large message sizes?

Any insights would be greatly appreciated. Thank you!

[acogoluegnes]

It is likely the consumer read all the messages in the stream, but there may extra information at the end of the stream and the offset lag is based on the offset of this piece of data.

Did you just publish messages to this stream, no stored offset for consumers or use of deduplication for publishers?

[hiimjako]

Messages are published by the AMQP client in C (not a stream client) to an exchange. The exchange is bind to a queue (which is out of scope here) and the stream specified above.
The publisher does not use any deduplication logic, and the consumer never publishes the offset.