SuperStream doesn't elect the single active consumer

[Gsantomaggio]

Describe the bug

The super stream doesn't elect the single active consumer when the consumers are restarted.
The consumers stop consuming and the status is always in waiting status. see the image:

Reproduction steps

1. Create super stream with ten partitions rabbitmq-streams add_super_stream invoices --partitions 10
2. pump it with a few thousand messages (70k is enough)
3. stop the producer
4. start ten instances of this java client:

    public static void main(String[] args) throws IOException {

        System.out.println("Connecting...");
        Address entryPoint = new Address("127.0.0.1", 5552);


        Environment environment = Environment.builder()
//                .host(entryPoint.host())
//                .port(entryPoint.port())
//                .username("test")
//                .password("test")
                .addressResolver(address -> entryPoint)
                .maxConsumersByConnection(1).
                build();
        String AppName = "reference";
        String stream = "invoices";

        AtomicInteger consumed = new AtomicInteger();
        Date start = new Date();
        for (int i = 0; i < 500; i++) {


            Map<String, Integer> consumedMap = new HashMap<>();
            Consumer consumer = environment.consumerBuilder()
                    .superStream("invoices")
                    .name("reference")
                    .offset(OffsetSpecification.first())
                    .singleActiveConsumer()
                    .messageHandler((context, message) -> {

                        if (consumedMap.containsKey(stream)) {
                            consumedMap.put(stream, consumedMap.get(stream) + 1);
                        } else {
                            consumedMap.put(stream, 1);
                        }

                        if (consumedMap.get(stream) % 10 == 0) {
                            Date end = new Date();
                            System.out.println("Stream: " + context.stream() + " - Consumed " + consumedMap.get(stream) + " - Time " + (end.getTime() - start.getTime()));
                        }


                        try {
                            Thread.sleep(ThreadLocalRandom.current().nextInt(200, 1000));
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }

                    }).build();

            try {
                Thread.sleep(60000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("Restarting");

            consumer.close();
        }

    }

5. Wait a couple of restarts
6. you will have the issue

Expected behavior

One single active consumer has to be active

Additional context

I noticed that the invoices-1 is usually the first portion to have problems.
The invoices-0 partition usually works
The other partitions, at some point, will have the same issue.

RabbitMQ 3.11.11
Java RabbitMQ Stream / Java 0.10.0-SNAPSHOT

[fabiorosa-sn]

We are currently facing the same problem with version 3.12.10 , Erlang 25.3.2.7.

It happened 2 in the ~ 3 months we are running with SuperStreams.

The only way for us to "fix it" is to scale down the consumer application to 0, and then scale up again.

[Gsantomaggio]

@fabiorosa-sn have you enabled stream_sac_coordinator_unblock_group? which stream client are you using?

[fabiorosa-sn]

yes, it's enabled.

we are using the java stream client version 0.15.0

[fabiorosa-sn]

Here is the result of the command

rabbitmqctl eval 'rabbit_stream_coordinator:sac_state(rabbit_stream_coordinator:state()).'

as requested by @Gsantomaggio

stream_coordinator_node-01.txt

[msvechla]

@Gsantomaggio This issue still exists, we are on RabbitMQ 3.12.14 Erlang 26.2.5.5

We are able to reproduce the issue with 5 active consumers and restarting all of them a couple of times.

Looking at the screenshots you can see that after some restarts one stream does not have any active consumer, with all consumers being in waiting state.

Can you give some more hints on how to diagnose this further? Thanks!

[mkuratczyk]

@msvechla the most helpful step towards solving this problem would be to provide the exact steps for reproducing the issue (even if one of the steps is non-deterministic, like "restart a few times"). Ideally you could use https://github.com/rabbitmq/rabbitmq-stream-perf-test so that we can easily follow the steps, but you can also use your own code, but provide it in a complete, executable format (a GH repo we can clone and build).

Please be precise: start with a fresh RabbitMQ, declare the super-stream (show us how you do it), then run stream-perf-test (or your app) - provide the exact command, etc.

If you can put together such a test case, please open a new issue with all the details.

[msvechla]

@mkuratczyk thanks. As we are currently not running this on our infrastructure, but on cloudamqp, we will also check with the support.

Thanks for providing the instructions, we will work on such a reproducible setup in case we can not find the root-cause there.

Just for my understanding: The scenario as shown in my screenshots above, as well as others in this issue here, where there are multiple connected consumers, but all of them are in waiting state, should never happen right?

Why is the RabbitMQ server not determining that there is no active consumer and attempting to elect a new one? Should we not see this in the RabbitMQ server logs somewhere?

[mkuratczyk]

Just for my understanding: The scenario as shown in my screenshots above, as well as others in this issue here, where there are multiple connected consumers, but all of them are in waiting state, should never happen right?

Yes, it should pick one of the available consumers.

Why is the RabbitMQ server not determining that there is no active consumer and attempting to elect a new one?

Well, that's what we need to understand and reproduction steps would help grately. :)

[jonnepmyra]

Hi, i can confirm that the issue still exists, running 3.13.7. This happened after a node restart (in a cluster of 3 rabbitmq nodes).

Don't know if it matters, but our clusters are relative big, with around 1000 superstreams.... so i guess there's ALOT of SAC-switching happening when we restart a rmq-node

[dukhaSlayer]

Can confirm, issue is still reproducible on 3.13.7

[michaelklishin]

@dukhaSlayer 3.13.x and 4.0.x are out of community support.

There were several stream SAC changes considered or adopted for 4.1.0, such as #13657 (merged) and #13671 (rejected).

Time to upgrade.

[Gsantomaggio]

There were several stream SAC changes considered or adopted for 4.1.0, such as #13657 (merged) and #13671 (rejected).

Please also follow this PR: #13672

[michaelklishin]

Thank you, @Gsantomaggio. That's indeed relevant, just hasn't been finished or shipped yet :)