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]

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 :(