Reusing of MQTT connection (clientId) results in error log "... Never happens."

[aretap]

This issue is slightly realated to GH-3685 (#3685).

I am trying to use one MQTT connection with a single clientId for subscriber as well as publisher channels.

Problem is that in MqttConnectOptions is the timeout specified in seconds while in org.eclipse.paho.client.mqttv3.MqttToken.waitForCompletion(long) is expected in milliseconds.

MqttConnectOptions.setConnectionTimeout(int seconds)
client.connect(this.connectionOptions).waitForCompletion(long milliseconds)

So calling:

Mqttv3ClientManager.start() {
  ...
  client.connect(this.connectionOptions).waitForCompletion(this.connectionOptions.getConnectionTimeout());
  ...
}

times out very quickly (default seems to be 30 seconds but they actually time out in 30 milliseconds) and an attempt to call the reconnect() method is made even though the connection is still becoming connected. That results in:

17:17:04,334 ERROR [main] o.s.i.m.c.Mqttv3ClientManager:110 MQTT client failed to connect. Never happens.
Connect already in progress (32110)
        at org.eclipse.paho.client.mqttv3.MqttAsyncClient.reconnect(MqttAsyncClient.java:1399)
        at org.springframework.integration.mqtt.core.Mqttv3ClientManager.start(Mqttv3ClientManager.java:107)
        at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:178)

This just prints the error message in log, but the connection seems to be established without further problems and the application works.

[artembilan]

Yeah... That's a bug: we must not use getConnectionTimeout() for waitForCompletion().

Possible better to introduce completionTimeout to the ClientManager API like we have in channel adapters.

Feel free to contribute the fix: https://github.com/spring-projects/spring-integration/blob/main/CONTRIBUTING.adoc !

[aretap]

I was thinking about just simply converting the getConnectionTimeout() from seconds to milliseconds before passing it to the waitForCompletion(long millis) method which, actually, waits for the connection to become estabilished to the server as specified in the MqttConnectOptions.setConnectionTimeout(int) method contract (JavaDoc) "... time interval the client will wait for the network connection to the MQTT server to be established". So it seems to me okay to use that timeout value in the waitForCompletion(long millis) method when connecting to the MQTT Broker server during the application startup.

	/**
	 * Sets the connection timeout value. This value, measured in seconds, defines
	 * the maximum time interval the client will wait for the network connection to
	 * the MQTT server to be established. The default timeout is 30 seconds. A value
	 * of 0 disables timeout processing meaning the client will wait until the
	 * network connection is made successfully or fails.
	 * 
	 * @param connectionTimeout
	 *            the timeout value, measured in seconds. It must be >0;
	 */
	public void setConnectionTimeout(int connectionTimeout) {
        ...
	}

Are you sure that introducing a new completionTimeout constant in the ClientManager API is the way to go?

[artembilan]

Well, that's probably OK as well.
What do you think about wrong usage of this property in the same class when we call client.disconnectForcibly() in the stop()?
In the end that one ends up on a waitForCompletion() as well.
Can we re-use the same getConnectionTimeout() for that part, too?

Same problem with Mqttv5ClientManager.

[aretap]

You are right that the diconnectForcibly(long) method is exactly the same use-case of the error. You are also right that the same use-case is in the Mqttv5ClientManager class. That makes it four problematic places now.

If I look at the MqttPahoMessageDrivenChannelAdapter class and how the getCompletionTimeout() method is used there for connecting/disconnecting and subscribing/unsubscribing, it seems to me legitimate to re-use the getConncetionTimeout() method for both, connecting and diconnecting to the MQTT server after the values are correctly converted to the target method time units.

[artembilan]

So, to make ourselves on the same page: do you suggest to fix both ClientManager impls to use that getConncetionTimeout() for all the use-case and with a proper converting to milliseconds?
On the other hand it sounds like you suggest to deprecate getCompletionTimeout() (and therefore disconnectCompletionTimeout as well) in favor of just getConncetionTimeout() in the AbstractMqttMessageDrivenChannelAdapter and in the and in the AbstractMqttMessageHandler?

Looks like in the Mqttv5PahoMessageDrivenChannelAdapter we do use a getCompletionTimeout() for disconnect as well though...
Therefore there was really a reason in those extra properties since in the end they don't rely on a conncetionTimeout in the Paho client.
So, don't we need instead just introduce a completionTimeout into a ClientManager API for safety and consistency with the Paho client?
I might agree about deprecating a disconnectCompletionTimeout since in the end that has to just a completionTimeout either way.
But about reusing conncetionTimeout for everything I'm in doubt yet...

[aretap]

I hear you when you look at it from the higher perspective than me in the beginning. Now it sounds reasonable to me to introduce a completionTimeout property to the ClientManager.

About deprecating the disconnectCompletionTimeout... I don't mind either way. The focus here should be on keeping the consistency.

[artembilan]

Sure! So, we go this route then:

1. Introduce completionTimeout into a ClientManager abstraction and fix inappropriate usages of the conncetionTimeout over there.
2. Deprecate a disconnectCompletionTimeout in favor of completionTimeout for consistency with other similar places.

Contribution is welcome!