Client-side offset tracking and old messages #151

staswilf · 2022-07-04T09:03:10Z

staswilf
Jul 4, 2022

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 · 2022-07-04T09:17:02Z

staswilf
Jul 4, 2022
Author

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

3 replies

acogoluegnes Jul 4, 2022
Maintainer

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 Jul 4, 2022
Author

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 Jul 4, 2022
Maintainer

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

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Client-side offset tracking and old messages #151

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{editor}}'s edit

{{editor}}'s edit

Uh oh!

Replies: 1 comment 3 replies

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{editor}}'s edit

{{editor}}'s edit

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Select a reply

Uh oh!

Client-side offset tracking and old messages #151

Uh oh!

Uh oh!

staswilf Jul 4, 2022

Replies: 1 comment · 3 replies

Uh oh!

staswilf Jul 4, 2022 Author

Uh oh!

Uh oh!

acogoluegnes Jul 4, 2022 Maintainer

Uh oh!

staswilf Jul 4, 2022 Author

Uh oh!

acogoluegnes Jul 4, 2022 Maintainer

staswilf
Jul 4, 2022

Replies: 1 comment 3 replies

staswilf
Jul 4, 2022
Author

acogoluegnes Jul 4, 2022
Maintainer

staswilf Jul 4, 2022
Author

acogoluegnes Jul 4, 2022
Maintainer