Storing streams offsets using Consumer#store(long) is not working (silently)

[aaltihami]

As per the blog https://blog.rabbitmq.com/posts/2021/09/rabbitmq-streams-offset-tracking/

"
The application code handles the tracking explicitly with Context#storeOffset() .
The Consumer#store(long) method is another possibility to store an offset
"

The Consumer#store(long) method is supposed to change the offset, However it doesn't store the new offset and there is no exception thrown.

environment
	.consumerBuilder()
	.stream(stream)
	.name("consumer_" + stream)
	.manualTrackingStrategy()
	.checkInterval(Duration.ofSeconds(2))
	.builder()
	.offset(OffsetSpecification.first())
	.messageHandler((context, message) -> {
		long messageContent = Long.parseLong(new String(message.getBodyAsBinary()));
		System.out.println(" message : " + messageContent);
		if (someCondition()) {
		    // moving the offset is not working 
		    context.consumer().store(OffsetSpecification.next().getOffset());
		}
	})
	.build();

[acogoluegnes]

The following line is wrong:

context().consumer().store(OffsetSpecification.next().getOffset());

It should be:

context().consumer().store(context.offset());

[aaltihami]

In the use case I'm working on I need to skip all of the stream messages if some condition is fulfilled, Is it possible?

[acogoluegnes]

You cannot "fast-forward" like this, you have to re-create the consumer.

If the number of messages to skip remains small (a few hundreds), it's OK to just skip them in the message handler.

[aaltihami]

Can such functionality be implemented "fast-forward"?

It might be helpful where the number of messages needed to be skipped is large.

And if it's doable, Where it should be implemented, rabbitmq server, rabbitmq Java client or both?

[acogoluegnes]

It could be a client-side feature, the library would take care of re-creating a subscription at the appropriate offset. This would be transparent to the application code.

I'll create a follow-up issue, I'd like to see the interest in such a feature, as it may not be trivial to implement.

[acogoluegnes]

Follow-up issue: #80.

[aaltihami]

That's great, thank you.

[acogoluegnes]

I will convert this issue to a GitHub discussion. Currently GitHub will automatically close and lock the issue even though your question will be transferred and responded to elsewhere. This is to let you know that we do not intend to ignore this but this is how the current GitHub conversion mechanism makes it seem for the users :(

[aaltihami]

Great, Thank you.