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();
    }
}