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