GH-3732 make mqttClient initialize lazily #3747

mipo256 · 2022-03-14T20:52:06Z

Fixing #3732 issue. Also simplified the Topic#equals() method.

artembilan

Please, find some review on the lines.
Thank you so far!

artembilan · 2022-03-15T14:08:06Z

...va/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java

@@ -140,7 +141,7 @@ protected MqttMessageConverter getConverter() {
 	}

 	@ManagedAttribute
-	public String[] getTopic() {
+	public String[] getTopicNames() {


Why did you rename the method?

I might be agree that your one has more sense, but for consistency with others we need to keep the old name.
See addTopic(), removeTopic().

Even if we call it getTopics(), we still need to keep the old one for backward compatibility.
Even if we deprecate it.

I mean the bad method name is not a strong justification to introduce a braking change.

I agree, you are right. I will leave the name of the method unchanged, thanks

artembilan · 2022-03-15T14:09:44Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

 		try {
 			this.mqttClient.unsubscribe(topics).waitForCompletion(getCompletionTimeout());
 			this.mqttClient.disconnect().waitForCompletion(getCompletionTimeout());
-		}
-		catch (MqttException ex) {
+		} catch (MqttException ex) {


That's not correct code style.

Please, run ./gradlew clean :spring-integration-mqtt:check to see all the Checkstyle violations.

Ok, sure, I am sorry for violating the project code style, I will polish the PR before merge. Thanks!

artembilan · 2022-03-15T14:11:23Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

@@ -187,33 +190,40 @@ protected void doStart() {
 	@Override
 	protected void doStop() {
 		this.topicLock.lock();
-		String[] topics = getTopic();
+		initializeMqttAsyncClientIfRequired();


That's something wrong in the stop() logic.
if there is not client initialized why would one wants to initialize it form stopping?

I think you are right, in general. There is clearly, from a logical perspective no need to initialize mqttClient in case we are done with this ChannelAdapter, so I agree.

But we still need to avoid NullPointerException, right? Falling with this exception will not be good, so maybe we can end up with the following scenario: we just check, whether mqttClient is not null, and if so, we will apply disconnection logic, otherwise we will just do nothing

artembilan · 2022-03-15T14:14:57Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

 			logger.error(ex, "Failed to close 'MqttAsyncClient'");
 		}
 	}

 	@Override
 	public void addTopic(String topic, int qos) {
 		this.topicLock.lock();
+		initializeMqttAsyncClientIfRequired();


That's not OK, too.
The addTopic() could be called from the Spring context initialization phase. Therefore it is too aggressive to make a low-level resource connection.

I believe the logic here must be similar to the one in the MqttPahoMessageDrivenChannelAdapter:

if (this.client != null && this.client.isConnected()) { this.client.subscribe(topic, qos); }

Ok, so I got the idea. We will just subscribe to the passed topic with passed QoS in case the mqttClient have being initialized, otherwise we will just maybe print a warning, just to notify the user.

@artembilan I think we should also omit the super.addTopic(topic, qos) invocation in the case of mqttClient have not been initialized, in order to not to get in trouble to return with getTopic() the name of the topic onto which we did not subscribed really.

mipo256

@artembilan I have adjusted the PR after your comments, can you, please, take a look again?

mipo256 · 2022-03-15T19:37:51Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

@@ -187,33 +190,40 @@ protected void doStart() {
 	@Override
 	protected void doStop() {
 		this.topicLock.lock();
-		String[] topics = getTopic();
+		initializeMqttAsyncClientIfRequired();


I think you are right, in general. There is clearly, from a logical perspective no need to initialize mqttClient in case we are done with this ChannelAdapter, so I agree.

But we still need to avoid NullPointerException, right? Falling with this exception will not be good, so maybe we can end up with the following scenario: we just check, whether mqttClient is not null, and if so, we will apply disconnection logic, otherwise we will just do nothing

mipo256 · 2022-03-15T19:38:32Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

 		try {
 			this.mqttClient.unsubscribe(topics).waitForCompletion(getCompletionTimeout());
 			this.mqttClient.disconnect().waitForCompletion(getCompletionTimeout());
-		}
-		catch (MqttException ex) {
+		} catch (MqttException ex) {


Ok, sure, I am sorry for violating the project code style, I will polish the PR before merge. Thanks!

mipo256 · 2022-03-15T19:49:21Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

 			logger.error(ex, "Failed to close 'MqttAsyncClient'");
 		}
 	}

 	@Override
 	public void addTopic(String topic, int qos) {
 		this.topicLock.lock();
+		initializeMqttAsyncClientIfRequired();


Ok, so I got the idea. We will just subscribe to the passed topic with passed QoS in case the mqttClient have being initialized, otherwise we will just maybe print a warning, just to notify the user.

@artembilan I think we should also omit the super.addTopic(topic, qos) invocation in the case of mqttClient have not been initialized, in order to not to get in trouble to return with getTopic() the name of the topic onto which we did not subscribed really.

artembilan

I also would like to see some simple unit test to verify the we don't have a reported in the issue NPE any more.

Thanks

artembilan · 2022-03-16T15:16:56Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

+				this.mqttClient.subscribe(topic, qos).waitForCompletion(getCompletionTimeout());
+				super.addTopic(topic, qos);
+			} else {
+				logger.warn(String.format("An attempt to add topic : '%s' with QoS : '%s' without having mqtt client initialized or connected, the topic will not be added!", topic, qos));


I don't think it is so bad to store topics (super.addTopic(topic, qos);) for the future use.
Something similar what we have so far in the MqttPahoMessageDrivenChannelAdapter.
So, please, change the logic over here to the old one and subscribe only if there is an mqttClient.

Well, ok, perhaps) I just retained a warning in case mqttClient is not initialized

artembilan · 2022-03-16T15:18:26Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

+		}
+	}
+
+	private void initializeMqttAsyncClient() {


This method becomes and obsolete right now, so better to restore the old code of the onInit()

That's right, I have removed it

mipo256 · 2022-03-17T15:39:21Z

@artembilan I have supplied the RP with 2 simple unit tests and squashed commits into one, can you please review?

Thanks

artembilan · 2022-03-17T15:53:26Z

squashed commits into one

That was not good, since now I have to review all your changes.
While without squashing there would be only need to take a look into fresh commits.

Not problem for now: jut want to give you a hint how it would be easier to collaborate in the future. 😄
We squash it in the end when we merge it upstream.

artembilan · 2022-03-17T15:57:43Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

+				this.mqttClient.unsubscribe(topics).waitForCompletion(getCompletionTimeout());
+				this.mqttClient.disconnect().waitForCompletion(getCompletionTimeout());
+			} else {
+				logger.warn("An attempt to stop mqtt client without having one initialized or connected");


I still want you to remove this warn.

The point is that component could not be started at all from the beginning for some reasons, but stop() is called when we close our application context.
Or the stop might be called via some centralized logic.
And the main feature of the stop is to be idempotent: nothing to stop - just exit silently!
So, such a warning would be a noise in logs in some target applications.

I agree, that make sense, I will remove this warning, thank you!

artembilan · 2022-03-17T15:58:40Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

+			if (mqttClient != null) {
+				this.mqttClient.close(true);
+			}
+		} catch (MqttException ex) {


Please, run gradle :spring-integration-mqtt:check to revise your Checkstyle violations and fix them, please.

Sorry, I have fixed all of the Checkstyle violations

artembilan · 2022-03-17T16:02:33Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

+			if (mqttClient != null && mqttClient.isConnected()) {
+				this.mqttClient.subscribe(topic, qos).waitForCompletion(getCompletionTimeout());
+			} else {
+				logger.warn(String.format("An attempt to add topic : '%s' with QoS : '%s' without having mqtt client initialized or connected", topic, qos));


This one has to be removed, too.
It is fully OK to add topics which could be subscribed later on, when start() is called.

artembilan · 2022-03-17T16:02:43Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

+			if (mqttClient != null && mqttClient.isConnected()) {
+				this.mqttClient.unsubscribe(topic).waitForCompletion(getCompletionTimeout());
+			} else {
+				logger.warn(String.format("An attempt to remove topics : '%s' without having mqtt client initialized or connected", Arrays.toString(topic)));


artembilan · 2022-03-17T16:04:26Z

.../org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapterTest.java

+class Mqttv5PahoMessageDrivenChannelAdapterTest {
+
+	@Test //GH-3732
+	public void testNoNpeIsNotThrownInCaseDoInitIsNotInvokedBeforeTopicAddition() {


How about just add these new tests into an existing Mqttv5BackToBackTests?

Sorry, I did not noticed this test class, I will move the test methods into Mqttv5BackToBackTests

…ding or removing topic

mipo256

I have fixed all of your suggestions and verified the code with Checkstyle, sorry for being annoying :)

mipo256 · 2022-03-18T10:03:05Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

+				this.mqttClient.unsubscribe(topics).waitForCompletion(getCompletionTimeout());
+				this.mqttClient.disconnect().waitForCompletion(getCompletionTimeout());
+			} else {
+				logger.warn("An attempt to stop mqtt client without having one initialized or connected");


I agree, that make sense, I will remove this warning, thank you!

mipo256 · 2022-03-18T10:08:01Z

.../org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapterTest.java

+class Mqttv5PahoMessageDrivenChannelAdapterTest {
+
+	@Test //GH-3732
+	public void testNoNpeIsNotThrownInCaseDoInitIsNotInvokedBeforeTopicAddition() {


Sorry, I did not noticed this test class, I will move the test methods into Mqttv5BackToBackTests

mipo256 · 2022-03-18T11:26:01Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

+			if (mqttClient != null) {
+				this.mqttClient.close(true);
+			}
+		} catch (MqttException ex) {


Sorry, I have fixed all of the Checkstyle violations

mipo256 · 2022-03-18T11:30:04Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

+			if (mqttClient != null && mqttClient.isConnected()) {
+				this.mqttClient.unsubscribe(topic).waitForCompletion(getCompletionTimeout());
+			} else {
+				logger.warn(String.format("An attempt to remove topics : '%s' without having mqtt client initialized or connected", Arrays.toString(topic)));


mipo256 · 2022-03-18T14:53:03Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

+			if (mqttClient != null && mqttClient.isConnected()) {
+				this.mqttClient.subscribe(topic, qos).waitForCompletion(getCompletionTimeout());
+			} else {
+				logger.warn(String.format("An attempt to add topic : '%s' with QoS : '%s' without having mqtt client initialized or connected", topic, qos));


artembilan

Pulling locally for the final review and possible merge...

artembilan · 2022-03-18T18:57:46Z

Merged and cherry-picked to 5.5.x as c1d29ab after some minor code clean up.

@mikhail2048 ,

thank you very much for the contribution; looking forward for more!

artembilan requested changes Mar 15, 2022

View reviewed changes

mipo256 commented Mar 15, 2022

View reviewed changes

mipo256 requested a review from artembilan March 15, 2022 20:52

artembilan requested changes Mar 16, 2022

View reviewed changes

mipo256 force-pushed the GH-3732 branch from 8d88fbf to 3a701d3 Compare March 17, 2022 15:36

mipo256 requested a review from artembilan March 17, 2022 15:37

artembilan requested changes Mar 17, 2022

View reviewed changes

spring-projectsGH-3732 add mqqtClient initialization check before ad…

08ade3b

…ding or removing topic

mipo256 force-pushed the GH-3732 branch from 3a701d3 to 08ade3b Compare March 18, 2022 15:04

mipo256 commented Mar 18, 2022

View reviewed changes

mipo256 requested a review from artembilan March 18, 2022 15:07

artembilan approved these changes Mar 18, 2022

View reviewed changes

artembilan closed this Mar 18, 2022

GH-3732 make mqttClient initialize lazily #3747

GH-3732 make mqttClient initialize lazily #3747

Conversation

mipo256 commented Mar 14, 2022

artembilan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mipo256 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

artembilan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mipo256 commented Mar 17, 2022

artembilan commented Mar 17, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mipo256 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

artembilan left a comment

Choose a reason for hiding this comment

artembilan commented Mar 18, 2022