rabbitmq client tries to reconnect while shutting down

[rastislavpapp]

Version

4.5.7

Context

While shutting down (e.q. when a quarkus app is restarted in dev mode), the client tries to reconnect to a queue, because io.vertx.rabbitmq.impl.QueueConsumerHandler#handleShutdownSignal calls a shutdownHandler defined on a queue, which is:

handler.setShutdownHandler(sig -> {
          restartConsumer(0, handler, options);
        });

so the queue starts to reconnect on shudown, which causes an exception, and in my case, second live connection to RabbitMQ server (and eventually a third one which I don't know where came from).

The exception:

2024-06-02 09:53:07,835 ERROR [io.ver.rab.imp.RabbitMQClientImpl] (vert.x-worker-thread-4) Exception whilst running connection stablished callback:  [Error Occurred After Shutdown]: java.lang.NullPointerException: Cannot invoke "io.smallrye.context.SmallRyeContextManager.defaultThreadContext()" because the return value of "io.smallrye.context.SmallRyeContextManagerProvider.getManager()" is null
        at io.smallrye.context.SmallRyeThreadContext.getCurrentThreadContextOrDefaultContexts(SmallRyeThreadContext.java:160)
        at io.smallrye.mutiny.context.DefaultContextPropagationInterceptor.getThreadContext(DefaultContextPropagationInterceptor.java:12)
        at io.smallrye.mutiny.context.BaseContextPropagationInterceptor.decorate(BaseContextPropagationInterceptor.java:24)
        at io.smallrye.mutiny.infrastructure.Infrastructure.decorate(Infrastructure.java:149)
        at io.smallrye.mutiny.groups.UniOnItem.call(UniOnItem.java:97)
        at io.smallrye.mutiny.Uni.call(Uni.java:434)
        at io.smallrye.reactive.messaging.rabbitmq.internals.IncomingRabbitMQChannel.lambda$createConsumer$5(IncomingRabbitMQChannel.java:129)
        at io.vertx.rabbitmq.impl.RabbitMQClientImpl.connectCallbackHandler(RabbitMQClientImpl.java:867)
        at io.vertx.rabbitmq.impl.RabbitMQClientImpl.connect(RabbitMQClientImpl.java:852)
        at io.vertx.rabbitmq.impl.RabbitMQClientImpl.lambda$tryConnect$36(RabbitMQClientImpl.java:763)
        at io.vertx.core.impl.ContextImpl.lambda$executeBlocking$1(ContextImpl.java:191)
        at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:279)
        at io.vertx.core.impl.ContextImpl.lambda$internalExecuteBlocking$2(ContextImpl.java:210)
        at io.vertx.core.impl.TaskQueue.run(TaskQueue.java:76)
        at org.jboss.threads.ContextHandler$1.runWith(ContextHandler.java:18)
        at org.jboss.threads.EnhancedQueueExecutor$Task.doRunWith(EnhancedQueueExecutor.java:2516)
        at org.jboss.threads.EnhancedQueueExecutor$Task.run(EnhancedQueueExecutor.java:2495)
        at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.run(EnhancedQueueExecutor.java:1521)
        at org.jboss.threads.DelegatingRunnable.run(DelegatingRunnable.java:11)
        at org.jboss.threads.ThreadLocalResettingRunnable.run(ThreadLocalResettingRunnable.java:11)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.base/java.lang.Thread.run(Thread.java:1583)

Do you have a reproducer?

No

Steps to reproduce

1. start quarkus app with a rabbitmq connection in dev mode
2. trigger restart
3. done, you have 3 connections and an exception

Extra

• env: quarkus 3.11.0, windows 11
• I've also created a bug in quarkus project (dev mode reload - rabbitmq - multiple rabbitmq connections quarkusio/quarkus#39433) but after further inspection I think the problem is in this client

[Yaytay]

The client will attempt to restart unless it has been closed.

[rastislavpapp]

So you are saying this is happening during consumer shutdown, and consumers nature is to reconnect after shutdown, and quarkus should be shutting down & starting the whole client instead?

[Yaytay]

Kindof, yes.
If the vertx rabbitmq client is configured to try to reconnect then it will keep doing so until the vertx rabbitmq client is closed, even if the vertx instance that it depends upon is shut down.
That does seem a bit unhelpful, but Vertx doesn't have a public API for knowing when (or whether) the vertx instance is shutdown.

I wrote a fix for this yesterday, but testing it is a bit awkward.
In the mean time the vertx rabbitmq client will either need to be shutdown, or not configured to re-attempt connections.

I think it's unusual to restart Vertx within a single process, which is why there aren't more people complaining about this (we do it all the time in tests, but they explicitly close the rabbitmq client).

[siewp]

I also stumbled across this problem in tests. IMHO the consumer should only restart if the shutdown signal is not coming from the application. That would be the same behaviour as for shutting down the whole connection. Would be a simple fix like this:

handler.setShutdownHandler(sig -> {
    if (!sig.isInitiatedByApplication()) {
        restartConsumer(0, handler, options);
    }
});

[Yaytay]

@siewp that works if the RabbitMQ connection is shutdown (and possibly I could save some code by using that), but in my testing it doesn't work when Vertx itself shuts down, I still need the VertxInternal#addCloseHook.