Unable to close reconnecting MQTT client

[Abhishek-Sharma007]

🐛 Bug Report

🔬 How To Reproduce

Steps to reproduce the behavior:

1. Implement mqtt client creation with autoReconnect=true, add connected and disconnected listener.
2. Introduce frequent network disturbances.
3. In each disturbances, close the current mqtt client and try to re-create a new one.
4. Observe in the thread dumps that multiple mqtt stale clients exists.

Code sample

public static synchronized Mqtt3Client getClient() {
            LOGGER.fine("Debug: MqttClientFactory.getClient called");
            Mqtt3Connect connOpts = Mqtt3Connect.builder()
                    .cleanSession(Boolean.TRUE)
                    .keepAlive(KEEP_ALIVE_TIME)
                    .build();
            activeClientId = String.valueOf(UUID.randomUUID());
            var builder = MqttClient.builder()
                    .useMqttVersion3()
                    .identifier(UUID.randomUUID().toString())
                    .transportConfig()
                    .mqttConnectTimeout(10, TimeUnit.SECONDS)
                    .socketConnectTimeout(10, TimeUnit.SECONDS)
                    .applyTransportConfig()
                    .automaticReconnectWithDefaultConfig()
                    .identifier(activeClientId)
                    .addConnectedListener(MqttClientFactory::handleConnection)
                    .addDisconnectedListener(MqttClientFactory::handleDisconnectionError)
                    .serverHost("localhost")
                    .serverPort(8883)
                    .simpleAuth()
                    .username("username")
                    .password("pass".getBytes())
                    .applySimpleAuth()
                    .sslWithDefaultConfig();
            mqtt3Client = builder.buildBlocking();
            mqtt3Client.connect(connOpts);
            LOGGER.log(Level.FINE, "Debug: MqttClientFactory.getClient returning mqtt3Client= {0}", mqtt3Client);

        return mqtt3Client;
    }

private static void handleConnection(MqttClientConnectedContext context) {
        LOGGER.log(Level.INFO, "MQTT Connection Established");
    }

    private static void handleDisconnectionError(MqttClientDisconnectedContext context) {
        String id = context.getClientConfig().getClientIdentifier().map(Objects::toString).orElse("");
        LOGGER.log(Level.INFO, "handleDisconnectionError:: id {0}", id);
        if (!id.equals(activeClientId)) {
            LOGGER.log(Level.INFO, "shutting down MQTT dead thread {0}", id);
            context.getReconnector().reconnect(false);
            return;
        }
        LOGGER.log(Level.WARNING,
                "MQTT connection disconnected due to : {0}",
                context.getCause().getMessage());
    }

    public static void shutdownConnection() {
        try {
            mqtt3Client.toBlocking().disconnect();
            mqtt3Client.getConfig().getDisconnectedListeners();
        } catch (Exception e) {
            LOGGER.log(Level.WARNING, "Failed to shutdown MQTT client);
        }
    }

1. Our application serves APIs to create and destroy MQTT client.
2. At max, the application can have single MQTT client.
3. Our internet connection is unstable and there are frequent connects & disconnects happening with MQTT client.
4. Observed that multiple MQTT client threads are alive, and disconnected Listenere is not getting invoked for older clients. So, it never reaches to the part where we have applied the workaround solution (How to turn off RECONNECT? #675)

Environment

Where are you running/using this client?

Hardware or Device - Both

What version of this client are you using? - 1.3.7

JVM version? - Java 21

Operating System? - Oracle Linux 9.6

Which MQTT protocol version is being used? - 3.1

Which MQTT broker (name and version)? - Mosquito 2.0.22

Screenshots

📈 Expected behavior

Older MQTT threads should also invoke connected/disconnected listeners when connectivity is restored/interrupted.
The above expectation should meet in order to have the workaround ((#675)) functioning.

📎 Additional context