How to properly recover from initial connection failures?

[pvaiko]

Hello,

We are running in a Kubernetes environment and when applications run up, the first thing they do is to try building an Environment using the builder. If RabbitMQ is not reachable at that moment (could happen for many reasons), the call throws an exception.

We then now periodically retry building the environment and when the broker becomes available, it connects to it and runs perfectly.

However, it seems like for every unsuccessful build() call, there is a multiThreadIoEventLoopGroup thread remaining in the runnable state, which, for longer downtimes may potentially cause resource problems.

What is the proper way to handle a non-reachable broker when build() is called?

Using 1.1 with RabbitMQ 4.1.

Here's a quick unit test, calling build() 100 times and showing 100 multiThreadIoEventLoopGroup threads remaining afterwards, just run it with no broker available:

import com.rabbitmq.stream.Environment;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertNull;

public class EnvironmentThreadsTest {

    private void countThreads() {
        ThreadGroup rootGroup = Thread.currentThread().getThreadGroup();
        ThreadGroup parentGroup;
        while ((parentGroup = rootGroup.getParent()) != null) {
            rootGroup = parentGroup;
        }
        Thread[] threads = new Thread[rootGroup.activeCount() * 2]; // Double to ensure we have enough space
        int actualCount = rootGroup.enumerate(threads, true);

        int eventLoopGroupThreadCount = 0;
        for (int i = 0; i < actualCount; i++) {
            Thread thread = threads[i];
            if (thread.getName().contains("multiThreadIoEventLoopGroup")) {
                eventLoopGroupThreadCount++;
            }
        }
        System.out.println("multiThreadIoEventLoopGroup thread count: " + eventLoopGroupThreadCount);
    }

    @Test
    public void testEnvironmentThreads() throws InterruptedException {
        Environment environment = null;
        for (int i = 0; i < 100; i++) {
            try {
                environment = Environment.builder().build();
                System.out.println("Environment created successfully.");
                break;
            } catch (Exception e) {
                assertNull(environment);
                System.out.println("Environment creation failed as expected since no RabbitMQ server is running");
            };
            Thread.sleep(100);
            countThreads();
        }
        Thread.sleep(1000);
        countThreads();
    }

    @Test
    public void testEnvironmentThreadsAutoClosable() throws InterruptedException {
        for (int i = 0; i < 100; i++) {
            try (Environment environment = Environment.builder().build()) {
                System.out.println("Environment created successfully.");
                break;
            } catch (Exception e) {
                System.out.println("Environment creation failed as expected since no RabbitMQ server is running");
            };
            Thread.sleep(100);
            countThreads();
        }
        Thread.sleep(1000);
        countThreads();
    }
}

[acogoluegnes]

I pushed a fix (#783) to close the Netty event loop group in case of initialization failure. You can also try the lazyInit flag: you'll be able to create the environment instance but each call to it will throw an exception until it manages to connect to the broker. You still have to properly close the environment, whether it is connected or not.

[pvaiko]

Thanks for the quick fix @acogoluegnes, will try the the lazyInit.