Client-side offset tracking and old messages

[staswilf]

Hi! I'm trying to track offset on the client side and sometimes get old messages.

I start like this:

offsetFromDB = readOffsetFromDB();
ConsumerBuilder builder = environment.consumerBuilder()
		.stream(config.getString("consumeQueue"))
		.subscriptionListener(subscriptionContext -> {
			subscriptionContext.offsetSpecification(OffsetSpecification.offset(offsetFromDB));
		}
		})
		.messageHandler(this);
Consumer cons = builder.build();

and then

@Override
public void handle(Context context, com.rabbitmq.stream.Message message) {
	if (offsetFromDB >= context.offset()) {
		// Message already processed
		return;
	}
	Message msg = Message.fromStreamMessage(
			message.getBodyAsBinary(),
			message.getProperties(),
			message.getApplicationProperties(),
			message.getMessageAnnotations(),
			context.offset());
	saveOffsetToDB(context.offset());
}

I expect that context.offset() always increments but sometimes I see batches of old messages (thousands) with smaller offsets passed to handle() and then suddenly everything gets OK. Am I missing something?

[staswilf]

Can I expect that sequential calls of context.offset() in the context of handle() return the same value?

[acogoluegnes]

Context#offset() should return the (unique) offset of the delivered Message. You're right, it should always return the same value in the context of a single to handle. It should increment in the next calls, under normal circumstances (no reconnection, etc).

The behavior you're observing is not normal. Your code seems correct. Which versions of Erlang/RabbitMQ/Java client are you using? Can you provide the server logs and the client logs?

Are you able to reproduce this behavior?

There can be duplicated messages e.g. when the connection dies and the client recovers it, but this should not happen if your offset tracking is reliable and is updated for each message.

[staswilf]

Thank you very much.
We have updated RMQ to 3.10.5 and the library to the latest version. There's no reliable reproduction; it happens sometimes on Android devices we have no direct control over. We'll try to collect detailed logs.

[acogoluegnes]

OK, thanks for the follow-up. You can also use the snapshot if that's possible for you.

https://rabbitmq.github.io/rabbitmq-stream-java-client/snapshot/htmlsingle/#dependencies
https://rabbitmq.github.io/rabbitmq-stream-java-client/snapshot/htmlsingle/#milestones-and-snapshots