[Suggestion] Stream - reset offset auto tracking for consumer

[unbearables]

RabbitMQ series

4.1.x

Operating system (distribution) used

AlmaLinux

How is RabbitMQ deployed?

RPM package

What would you like to suggest for a future version of RabbitMQ?

It would be useful to reset stream server-side offset auto tracking for a given consumer. I am aware that we can use manual offset tracking or we can change name of the consumer, but that is a messy workaround. It would be nice to leverage RabbitMQ offset auto tracking.

Changing tracking strategy to manual after auto tracking (consumer with same name) is also not possible as it is denied with message Requested offset specification OffsetSpecification{type=4, offset=11} not used in favor of stored offset found for reference app-test - this was tested via Java amqp-client 5.22.0 and stream-client 0.18.0.

Usecase for resetting an offset for consumer is rereading a stream without changing consumer's name + leaving the stream data intact. We have application, which uses partial data from stream, but afterwards needs more data from the stream, where rereading a stream is crucial. Secondary usecase is to clear offset tracking on long-lasting stream.

[acogoluegnes]

Would you mind experimenting with the Client#storeOffset(String, String, long) and Client#queryOffset(String, String) methods?

You need to create a Client instance along the Environment, but that should be OK for development. You can then store offset 0 for the consumer (same stream and reference/name), make sure it has been written (by polling with queryOffset) and then start the consumer. We could then discuss a proper API if the experiment addresses your use case.

[acogoluegnes]

There is also the subscription listener callback: it lets you compute the offset the subscription should start from. The result of the callback overrides the offset specification set on the consumer creation and the server-side stored offset (if any).

[unbearables]

I tested both your proposals and only Client#storeOffset(String, String, long) worked - I stopped consumer, created new Client, called storeOffset with offset = 0, closed the client and successfully reread the whole stream.

Second option via subscription listener did not work for me. It logs:

INFO c.r.stream.impl.ConsumersCoordinator.add - Computed offset specification OffsetSpecification{type=1, offset=-1}, offset specification used after subscription listener OffsetSpecification{type=4, offset=1}

so clearly it picked it up, but the stream was not reread.

It is kind of awkward to construct a new Client via ClientParameters - is it possible to utilize Environment somehow?

Also as this offset reset is an edge case - is it possible to do it via rabbitmq-streams CLI?

Nevertheless, thank you for your solution.

[acogoluegnes]

It is kind of awkward to construct a new Client via ClientParameters - is it possible to utilize Environment somehow?

Sure, we'll add a Environment#storeOffset(String, String, long).

Also as this offset reset is an edge case - is it possible to do it via rabbitmq-streams CLI?

Makes sense, something like rabbitmq-streams reset-offset my-stream my-consumer-name.

Second option via subscription listener did not work for me.

Note it does not change the stored offset: the value set in the subscription context has the final word and is used for the subscribe. As said in the documentation, it is a way to use an external store to store offsets.

I think the library did not find any stored offset (Computed offset specification OffsetSpecification{type=1, offset=-1}, type 1 is FIRST, so it is likely it used the offset specification set at the consumer creation). But it seems it picked up the value you set (offset = 1). Could it be the stream is empty?

Would you mind upgrading the library to 1.1.0 and give another try?

This is the current code, the offset set in the subscription context should be used.

[unbearables]

I created a small JBang util to reset stream offset so far. Run with jbang rabbitmq_stream_change_offset.java --host={HOST} --user={USER} --ref={REF} --stream={STREAM} --offset={DESIRED_OFFSET}. So far I am using com.rabbitmq:stream-client:0.18.0 as I have some netty package problem when I use 1.1.0 (java.lang.ClassNotFoundException: io.netty.channel.MultiThreadIoEventLoopGroup).

///usr/bin/env jbang "$0" "$@" ; exit $?
//DEPS info.picocli:picocli:4.6.3
//DEPS com.rabbitmq:stream-client:0.18.0
//DEPS org.apache.logging.log4j:log4j-api:2.20.0
//DEPS org.apache.logging.log4j:log4j-core:2.20.0
//DEPS org.apache.logging.log4j:log4j-slf4j-impl:2.20.0


import picocli.CommandLine;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;

import java.util.concurrent.Callable;

import com.rabbitmq.stream.impl.Client;

@Command(name = "rabbitmq_stream_change_offset", mixinStandardHelpOptions = true, version = "rabbitmq_stream_change_offset 0.1",
        description = "rabbitmq_stream_change_offset made with jbang")
class rabbitmq_stream_change_offset implements Callable<Integer> {

    @Option(names = { "--host" }, description = "RabbitMQ host", arity = "1")
    private String host;

    @Option(names = { "--port" }, description = "RabbitMQ stream port", defaultValue = "5552")
    private int port;

    @Option(names = { "--user" }, description = "RabbitMQ user username", arity = "1")
    private String username;

    @Option(names = { "--ref" }, description = "Consumer reference for reset (usually app name)", arity = "1")
    private String reference; // stream reference ~ app name

    @Option(names = { "--stream" }, description = "RabbitMQ stream to reset", arity = "1")
    private String stream;

    @Option(names = { "--offset" }, description = "New stream offset to set", defaultValue = "0")
    private long offset;

    public static void main(String... args) {
        int exitCode = new CommandLine(new rabbitmq_stream_change_offset()).execute(args);
        System.exit(exitCode);
    }

    @Override
    public Integer call() throws Exception {
        System.out.println("Given config for stream offset change:");
        System.out.println(" -- RabbitMQ     = " + host + ":" + port);
        System.out.println(" -- stream       = " + stream);
        System.out.println(" -- consumer ref = " + reference);
        System.out.println(" -- offset       = " + offset);
        System.out.println("\n");

        final String password = new String(System.console().readPassword("Password for user '" + username + "': "));
        if (password.isBlank()) {
            System.out.println("No password given");
            return 1;
        }

        System.out.println("Creating new client");
        final Client client = new Client(
            new Client.ClientParameters()
                .host(host)
                .port(port)
                .username(username)
                .password(password)
        );

        final Client.QueryOffsetResponse queryOffsetResponse = client.queryOffset(reference, stream);
        if (queryOffsetResponse.getResponseCode() != 1) {
            System.out.println("Could not query offset - responseCode=" + queryOffsetResponse.getResponseCode());
            return 1;
        }

        System.out.println("Changing offset from " + queryOffsetResponse.getOffset() + " to " + offset);
        if (queryOffsetResponse.getOffset() == offset) {
            System.out.println("No need to change offset");
            return 0;
        }

        client.storeOffset(reference, stream, offset);
        System.out.println("Closing client");
        client.close();
        System.out.println("Done");
        return 0;
    }
}

[acogoluegnes]

So far I am using com.rabbitmq:stream-client:0.18.0 as I have some netty package problem when I use 1.1.0 (java.lang.ClassNotFoundException: io.netty.channel.MultiThreadIoEventLoopGroup).

We bumped Netty to 4.2 in 0.24.0, so you should be able to use 0.23.0. I can see if it possible to fall back to Netty 4.1 if Netty 4.2 is not available. Why do you have to stick to Netty 4.1?

[acogoluegnes]

@unbearables I added the Environment#store(String,String,long) method, can you give it a try with the latest snapshot? Don't forget to declare the snapshot repository, as explained in the documentation.

The client library uses Netty 4.2 only if it is available, so there should be no version conflict now.