Nullpointerexception in spring-integration-mqtt adapter while reconnecting

[nitinsh99]

In what version(s) of Spring Integration are you seeing this issue?
5.5.3

Describe the bug

15:41:23.620 scheduler2 ERROR .s.s.s.TaskUtils$LoggingErrorHandler| Unexpected error occurred in scheduled task requestId=LnaKMnE6
java.lang.NullPointerException: null
	at org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter.connectAndSubscribe(MqttPahoMessageDrivenChannelAdapter.java:314)
	at org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter.lambda$scheduleReconnect$2(MqttPahoMessageDrivenChannelAdapter.java:364)
	at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1130)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:630)
	at java.base/java.lang.Thread.run(Thread.java:832)

A clear and concise description of what the bug is.

This line is throwing NullPointerException inside

this.client.disconnectForcibly(this.disconnectCompletionTimeout);

To Reproduce
Write a simple unit test that makes adapter fails to connect to a topic.. This is the order of the exceptions

15:57:29.009 Thread-31 ERROR .MqttPahoMessageDrivenChannelAdapter| Error connecting or subscribing to [publish/*/*] 
org.eclipse.paho.client.mqttv3.MqttException: MqttException
	at org.eclipse.paho.client.mqttv3.internal.ExceptionHelper.createMqttException(ExceptionHelper.java:38)
	at org.eclipse.paho.client.mqttv3.internal.ClientComms$ConnectBG.run(ClientComms.java:738)
	at java.base/java.lang.Thread.run(Thread.java:832)
Caused by: java.io.IOException: WebSocket Response header: Incorrect connection header
	at org.eclipse.paho.client.mqttv3.internal.websocket.WebSocketHandshake.receiveHandshakeResponse(WebSocketHandshake.java:157)
	at org.eclipse.paho.client.mqttv3.internal.websocket.WebSocketHandshake.execute(WebSocketHandshake.java:80)
	at org.eclipse.paho.client.mqttv3.internal.websocket.WebSocketSecureNetworkModule.start(WebSocketSecureNetworkModule.java:65)
	at org.eclipse.paho.client.mqttv3.internal.ClientComms$ConnectBG.run(ClientComms.java:724)
	... 1 common frames omitted

then

15:57:59.016 Thread-31 ERROR .MqttPahoMessageDrivenChannelAdapter| Exception while connecting and subscribing, retrying 
org.eclipse.paho.client.mqttv3.MqttException: MqttException
	at org.eclipse.paho.client.mqttv3.internal.ExceptionHelper.createMqttException(ExceptionHelper.java:38)
	at org.eclipse.paho.client.mqttv3.internal.ClientComms$ConnectBG.run(ClientComms.java:738)
	at java.base/java.lang.Thread.run(Thread.java:832)
Caused by: java.io.IOException: WebSocket Response header: Incorrect connection header
	at org.eclipse.paho.client.mqttv3.internal.websocket.WebSocketHandshake.receiveHandshakeResponse(WebSocketHandshake.java:157)
	at org.eclipse.paho.client.mqttv3.internal.websocket.WebSocketHandshake.execute(WebSocketHandshake.java:80)
	at org.eclipse.paho.client.mqttv3.internal.websocket.WebSocketSecureNetworkModule.start(WebSocketSecureNetworkModule.java:65)
	at org.eclipse.paho.client.mqttv3.internal.ClientComms$ConnectBG.run(ClientComms.java:724)
	... 1 common frames omitted

15:58:09.079 scheduler1 ERROR .MqttPahoMessageDrivenChannelAdapter| Exception while unsubscribing 
org.eclipse.paho.client.mqttv3.MqttException: Client is not connected
	at org.eclipse.paho.client.mqttv3.internal.ExceptionHelper.createMqttException(ExceptionHelper.java:31)
	at org.eclipse.paho.client.mqttv3.internal.ClientComms.sendNoWait(ClientComms.java:205)
	at org.eclipse.paho.client.mqttv3.MqttAsyncClient.unsubscribe(MqttAsyncClient.java:1244)
	at org.eclipse.paho.client.mqttv3.MqttClient.unsubscribe(MqttClient.java:551)
	at org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter.doStop(MqttPahoMessageDrivenChannelAdapter.java:223)
	at org.springframework.integration.en

then

15:58:39.083 scheduler1 ERROR .MqttPahoMessageDrivenChannelAdapter| Error connecting or subscribing to [publish/*/*] 
org.eclipse.paho.client.mqttv3.MqttException: MqttException
	at org.eclipse.paho.client.mqttv3.internal.ExceptionHelper.createMqttException(ExceptionHelper.java:38)
	at org.eclipse.paho.client.mqttv3.internal.ClientComms$ConnectBG.run(ClientComms.java:738)
	at java.base/java.lang.Thread.run(Thread.java:832)
Caused by: java.io.IOException: WebSocket Response header: Incorrect connection header
	at org.eclipse.paho.client.mqttv3.internal.websocket.WebSocketHandshake.receiveHandshakeResponse(WebSocketHandshake.java:157)
	at org.eclipse.paho.client.mqttv3.internal.websocket.WebSocketHandshake.execute(WebSocketHandshake.java:80)
	at org.eclipse.paho.client.mqttv3.internal.websocket.WebSocketSecureNetworkModule.start(WebSocketSecureNetworkModule.java:65)
	at org.eclipse.paho.client.mqttv3.internal.ClientComms$ConnectBG.run(ClientComms.java:724)
	... 1 common frames omitted

then the NPE

java.lang.NullPointerException: null
	at org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter.connectAndSubscribe(MqttPahoMessageDrivenChannelAdapter.java:314)
	at org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter.lambda$scheduleReconnect$2(MqttPahoMessageDrivenChannelAdapter.java:364)
	at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1130)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:630)
	at java.base/java.lang.Thread.run(Thread.java:832)

Steps to reproduce the behavior.

Expected behavior

Library should never throw NPE. Looks like code is setting client to null and during reconnect using a null client to reconnect

A clear and concise description of what you expected to happen.

Sample

A link to a GitHub repository with a minimal, reproducible sample.

Reports that include a sample will take priority over reports that do not.
At times, we may require a sample, so it is good to try and include a sample up front.

[artembilan]

Can you share, please, that simple unit test to reproduce?
I don't see how is that possible to have the client property reset to null in between this.client = this.clientFactory.getClientInstance(getUrl(), getClientId()); and this.client.disconnectForcibly(this.disconnectCompletionTimeout); in that connectAndSubscribe() since it is synchronized... As well as all other methods in this class which deal with this.client. So, I'm not sure how is that possible to break that thread barrier in normal program flow.

[nitinsh99]

@artembilan .. I digged a bit down on what's happening.. So I call

            flowContext
                    .getRegistry()
                    .get("id")
                    .destroy();

on MqttConnectionFailedEvent.. So what is happening is when the first connection fails, a MqttConnectionFailedEvent is fired then my logic destroys registration which in turn destroys all the bean in the registration hierarchy (including the client I presume ) and the adapter attempts reconnects since the reconnect future is already schedule and ends up causing NPE..

So, apart from NPE, another issue that I am struggling is how to make the adapater not to schedule a reconnect future? From the code it's looks like its not configurable and one retry is attempted regardles.. Please correct me.. cc: @garyrussell

My end goal is to simply remove the flow registration when the client fails to connect to broker

[artembilan]

Sorry, this flowContext makes things over here more complicated.
Please, share that simple unit test you have mentioned in the original post.
I really would like to understand how this NPE happens from the code perspective.
The connectAndSubscribe() does create a fresh client instance in its synchronized state. That is fully not clear how the client variable can be reset in between since all other methods are still synchronized therefore they wait for each other even if they are called concurrently.
So, the mentioned simple unit test would be great, because I might be wrong with my assumption and I really may miss something else since NPE has a fact to appear on your side.

Thanks for understanding.

[nitinsh99]

Ok.. I am working on writing a unit test for you but meanwhile if you get a chance, it's fairly easy to replicate...

Basically

@Autowired flowContext
 
final IntegrationFlow flow = IntegrationFlows
        .from(adapter)
        .handle((Message<?> message) -> this.handleMessage(id, message))
        .get();          
        
flowContext
        .registration(flow)
        .id("id")
        .register();
        
@Bean
public ApplicationListener<MqttConnectionFailedEvent> connectionFailureEventListener() {
    return (event) -> {
        final String id = getFlowContextRegistrationIdFromMqttEvent(event);
        flowContext
                .getRegistry()
                .get(id)
                .destroy(); <--- the reason client is becoming null despite a syncronised is
 because this destroys the registration bean hierarchy and there is a race condition
 b/w reconnect attempt and destroy method destroying the  client instance
    };
}
      
        

[artembilan]

OK. I was able to reproduce an issue with this unit-test (based on the framework test environment):

	@Test
	public void testNoNPEOnReconnectAndStopRaceCondition() throws Exception {
		final IMqttClient client = mock(IMqttClient.class);
		MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER);
		adapter.setRecoveryInterval(10);

		willThrow(MqttException.class)
				.given(client)
				.subscribe(any(),ArgumentMatchers.<int[]>any());

		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
		AtomicReference<Throwable> reconnectException = new AtomicReference<>();
		CountDownLatch reconnectExceptionLatch = new CountDownLatch(1);
		taskScheduler.setErrorHandler(ex -> {
			reconnectException.set(ex);
			reconnectExceptionLatch.countDown();
		});
		taskScheduler.initialize();
		adapter.setTaskScheduler(taskScheduler);

		adapter.setApplicationEventPublisher(event -> {
			if (event instanceof MqttConnectionFailedEvent) {
				adapter.destroy();
			}
		});
		adapter.start();

		assertThat(reconnectExceptionLatch.await(10, TimeUnit.SECONDS)).isTrue();
		assertThat(reconnectException.get())
				.isNotNull()
				.isInstanceOf(MqttException.class);

		taskScheduler.destroy();
	}

(I believe the MqttPahoMessageDrivenChannelAdapter instance could be any).

The main point of this test that I call (as you do) an adapter.destroy() within the same thread as MqttConnectionFailedEvent emitted from that connectAndSubscribe(), so we are good to go to the doStop() and null the client property and therefore it leads to NPE on the next line in that connectAndSubscribe() which is this.client.disconnectForcibly(this.disconnectCompletionTimeout);.
The test result without any code changes is like this:

2021-09-09 13:06:04,509 ERROR [main] [org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter] - Error connecting or subscribing to [foo]
org.eclipse.paho.client.mqttv3.MqttException: MqttException
2021-09-09 13:06:04,557 ERROR [main] [org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter] - Exception while connecting and subscribing, retrying
org.eclipse.paho.client.mqttv3.MqttException: MqttException
2021-09-09 13:06:04,583 ERROR [ThreadPoolTaskScheduler-1] [org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter] - Error connecting or subscribing to [foo]
org.eclipse.paho.client.mqttv3.MqttException: MqttException

java.lang.AssertionError: 
Expecting actual throwable to be an instance of:
  org.eclipse.paho.client.mqttv3.MqttException
but was:
  java.lang.NullPointerException
	at org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter.connectAndSubscribe(MqttPahoMessageDrivenChannelAdapter.java:314)
	at org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter.lambda$scheduleReconnect$2(MqttPahoMessageDrivenChannelAdapter.java:364)

Same NPE.
It is just because we are free to call any synchronized from the synchronized if we are withing the same execution flow on the same thread so far.

So, thank you for the report and stay tuned for the fix!