Reconnection / retry error with Paho MQTT 1.2 leads to ever increasing thread usage

[michael-simons]

As a follow up to my tweet.

Find attached a simple Spring Integration MQTT application (mqtt-paho-client-threading-bug.zip). It's a Spring Boot app, just run

./mvnw clean spring-boot:run

You'll find the app trying to reach a MQTT broker on tcp://localhost:45678 (I chose this port to make sure there is no broker answering).

The app will fail, print the MQTT Paho exception and continue.

I'll added a scheduled task that prints the number of live threads. As you see they'll increse by 2 threads after each failed connection retry.

This happens during startup as well as during runtime, for example if the broker goes away in between.

A downgrade to Paho 1.1.1 as suggested in eclipse-paho/paho.mqtt.java#402 respectivly SebaDro/SOS@8b9f09e helps, for example like this:

    <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-mqtt</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.eclipse.paho</groupId>
                    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.eclipse.paho</groupId>
            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
            <version>1.1.1</version>
            <scope>compile</scope>
        </dependency>

With Paho 1.1.1 the number of threads stays constant.

[hbergmey]

Unfortunately I cannot go back to 1.1.1 as then my client connections become unstable. As a matter of fact, I do not receive any further messages after the initial topic report. Paho 1.2.0 fixed this strange behavior.

Somehow we need to fix the thread shutdown issue.

[garyrussell]

They really need to fix the client; we can look at closing/creating a new client each time, but that just seems wrong.

[hbergmey]

Looking into the Paho Code, I see that the MqttAsyncClient can be setup to use an explicitly given ThreadPool like so:

new MqttAsyncClient(url, clientId, persistence, new TimerPingSender(), Executors.newScheduledThreadPool(10))

Do you see a chance to explicitly kill this threadpool on app/container shutdown?

Neither @PreDestroy nor finalize() cause my close() function with mqttClient.disconnectForcibly() to be called.

[garyrussell]

My mistake; it's our bug; we do create a new client on each attempt; we just don't clean up the old one.

[hbergmey]

I join the "mea culpa" chant: My mistake was the @PreDestroy annotation at the wrong class. Obviously it only works in a Spring service.

I am now calling the following in a @PreDestroy hook

@PreDestroy
public void close() {
   IMqttToken disconnectToken = mqttClient.disconnect();
   disconnectToken.waitForCompletion(10000);
   mqttClient.close(true);
   mqttClient = null;
}

[garyrussell]

I tested your app against the snapshot locally and there is no more leak.

Releases that will include this fix are scheduled for next Wednesday.

[michael-simons]

Wow, awesome reaction. Thanks a lot!

[garyrussell]

Need similar logic in the outbound adapter.

[jmax01]

Thank you and for reference, the details of what we experienced: https://gitter.im/spring-projects/spring-integration?at=5bbd2483bbdc0b250512e7ce

Need similar logic in the outbound adapter.