Stream client not always recovering after a server restart

[bvrnAXI]

Hello,

rabbitmq version: 3.13.7
stream-client version: 0.17.0

RabbitMQ server (one node) and java application running in docker on the same server.

we are experiencing some unstable behavior in the connection recovery of the stream-client.
After normal startup, our application produces and consumes messages to and from a rabbitmq super stream. When the rabbitmq server reboots, the streams always become available again at the server side.
Sometimes the client connection, producer and consumers come back up after some time and our application can continue without having to reboot itself, but at other occasions the connection and/or producer and/or consumers don't work anymore. Sometimes it seems the producer still sends messages but they don't appear in the streams. In the application logging messages like
com.rabbitmq.stream.impl.OffsetTrackingCoordinator: Error while flushing tracker: Not possible to query offset for consumer CmmnEventConsumer on stream CmmnEvents-2 for now, consumer status is NOT_AVAILABLE appear. In the rabbitmq server log sometimes the error in attachment appears, sometimes not and still, afterwards sometimes everything works again and sometimes not.
rabbit_error.txt

There is no pattern in when this happens. It seems (but can be random) that when the setup is done with TLS, the problems occur more often, but it also sometimes works and also without TLS these problems can occur.

-> some code snippets for environment, stream, consumer, producer creation

EnvironmentBuilder environmentBuilder = Environment.builder().uri("rabbitmq-stream://" + rabbitMQHost + ":" + rabbitMQStreamPort);

environment.streamCreator() .name(SUPER_STREAM_NAME) .maxAge(Duration.ofDays(30)) .maxSegmentSizeBytes(ByteCapacity.MB(500)) .superStream() .partitions(8) .creator() .create();

environment.producerBuilder().superStream(SUPER_STREAM_NAME) .routing(msg -> msg.getProperties().getMessageIdAsString()) .producerBuilder() .build()

environment.consumerBuilder() .superStream(SUPER_STREAM_NAME) .name(CONSUMER_NAME) .singleActiveConsumer() .autoTrackingStrategy() .builder() .messageHandler(messageHandler::handle) .build()

Any thoughts or ideas on this situation are appreciated.

Regards,
Ben

[acogoluegnes]

Can you share a standalone project with instructions that reproduces the problem? Thanks.

[bvrnAXI]

I have some extra information regarding this problem: It seems that the number of threads available for the ScheduledExecutorService of the StreamEnvironment has an influence on the recovery stability.
We have a superstream with 8 partitions and create 2 single active comsumers. When not defining a ScheduledExecutorService ourselves, the client will create one that uses as many threads as available processors on the system. On a system with 12 cpu's the recovery always succeeds, on a system with 2 cpu's it almost always fails.
When defining a ScheduledExecutorService with 12 or 16 threads on the StreamEnvironment, the recovery also succeeds on the system with 2 cpu's.

(rabbitmq version 4.0.3 and stream client version 0.21.0)

Any ideas on this behavior?

Thanks,
Ben

[acogoluegnes]

We released 0.22.0 yesterday with some improvements in recovery, can you give it a try?

As suggested above, a standalone project that reproduces the issue (on any machine, whatever the number of CPUs) would be useful to investigate the issue.

[bvrnAXI]

Yes, the latest release fixes our recovery problems !
Thank you very much. Would have saved us a lot of time if this was released last week though ;)

Regards,
Ben

[michaelklishin]

@bvrnAXI you could have contributed a fix earlier or at least asked for a new release.

These tongue-in-cheek demands are not appreciated. No one in the open source community owes you anything. Contribute or be thankful for getting this software, ongoing maintenance and guidance from the core team for free.