GH-3454: From MQTT conversion error - to error ch (#3456)

* GH-3454: From MQTT conversion error - to error ch

Fixes #3454

The message converter may return null when we try to covert from the
MQTT message.
The thrown exception may also reset the client connect.

* Fix `MqttPahoMessageDrivenChannelAdapter` to catch any conversion errors
(including `null` result) and try to send an `ErrorMessage` with that info
into the provided `errorChannel`.
Otherwise re-throw it as as

**Cherry-pick to `5.4.x` & `5.3.x`**

* * Apply review language-specific changes

# Conflicts:
#	spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java
#	spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java
#	src/reference/asciidoc/mqtt.adoc

Author: artembilan

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -37,6 +37,8 @@
 import org.springframework.integration.mqtt.event.MqttSubscribedEvent;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.converter.MessageConversionException;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.Assert;
 
 /**
@@ -169,7 +171,7 @@ protected void doStart() {
 		}
 		catch (Exception e) {
 			logger.error("Exception while connecting and subscribing, retrying", e);
-			this.scheduleReconnect();
+			scheduleReconnect();
 		}
 	}
 
@@ -181,7 +183,7 @@ protected synchronized void doStop() {
 			try {
 				if (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_ALWAYS)
 						|| (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_CLEAN)
-								&& this.cleanSession)) {
+						&& this.cleanSession)) {
 
 					this.client.unsubscribe(getTopic());
 				}
@@ -237,7 +239,7 @@ public void removeTopic(String... topic) {
 			super.removeTopic(topic);
 		}
 		catch (MqttException e) {
-			throw new MessagingException("Failed to unsubscribe from topic " + Arrays.asList(topic), e);
+			throw new MessagingException("Failed to unsubscribe from topic(s) " + Arrays.toString(topic), e);
 		}
 		finally {
 			this.topicLock.unlock();
@@ -365,14 +367,47 @@ public synchronized void connectionLost(Throwable cause) {
 
 	@Override
 	public void messageArrived(String topic, MqttMessage mqttMessage) {
-		Message<?> message = this.getConverter().toMessage(topic, mqttMessage);
+		Message<?> message = toMessage(topic, mqttMessage);
+		if (message != null) {
+			try {
+				sendMessage(message);
+			}
+			catch (RuntimeException e) {
+				logger.error("Unhandled exception for " + message.toString(), e);
+				throw e;
+			}
+		}
+	}
+
+	private Message<?> toMessage(String topic, MqttMessage mqttMessage) {
+		Message<?> message = null;
+		RuntimeException conversionError = null;
 		try {
-			sendMessage(message);
+			message = getConverter().toMessage(topic, mqttMessage);
 		}
-		catch (RuntimeException e) {
-			logger.error("Unhandled exception for " + message.toString(), e);
-			throw e;
+		catch (RuntimeException ex) {
+			conversionError = ex;
+		}
+
+		if (message == null && conversionError == null) {
+			conversionError = new IllegalStateException("'MqttMessageConverter' returned 'null'");
+		}
+
+		if (conversionError != null) {
+			GenericMessage<MqttMessage> failedMessage = new GenericMessage<>(mqttMessage);
+			if (!sendErrorMessageIfNecessary(failedMessage, conversionError)) {
+				MessageConversionException conversionException;
+				if (conversionError instanceof MessageConversionException) {
+					conversionException = (MessageConversionException) conversionError;
+				}
+				else {
+					conversionException = new MessageConversionException(failedMessage,
+							"Failed to convert from MQTT Message", conversionError);
+				}
+				throw conversionException;
+			}
 		}
+		return message;
 	}
 
 	@Override

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -85,9 +85,12 @@
 import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
 import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
 import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
+import org.springframework.integration.mqtt.support.MqttMessageConverter;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -102,7 +105,7 @@
  */
 public class MqttAdapterTests {
 
-	private IMqttToken alwaysComplete;
+	private final IMqttToken alwaysComplete;
 
 	{
 		ProxyFactoryBean pfb = new ProxyFactoryBean();
@@ -194,12 +197,12 @@ public void testOutboundOptionsApplied() throws Exception {
 			return deliveryToken;
 		}).given(client).publish(anyString(), any(MqttMessage.class));
 
-		handler.handleMessage(new GenericMessage<String>("Hello, world!"));
+		handler.handleMessage(new GenericMessage<>("Hello, world!"));
 
 		verify(client, times(1)).connect(any(MqttConnectOptions.class));
 		assertThat(connectCalled.get()).isTrue();
 		AtomicReference<Object> failed = new AtomicReference<>();
-		handler.setApplicationEventPublisher(event -> failed.set(event));
+		handler.setApplicationEventPublisher(failed::set);
 		handler.connectionLost(new IllegalStateException());
 		assertThat(failed.get()).isInstanceOf(MqttConnectionFailedEvent.class);
 		handler.stop();
@@ -254,7 +257,7 @@ public void testInboundOptionsApplied() throws Exception {
 			return null;
 		}).given(client).connect(any(MqttConnectOptions.class));
 
-		final AtomicReference<MqttCallback> callback = new AtomicReference<MqttCallback>();
+		final AtomicReference<MqttCallback> callback = new AtomicReference<>();
 		willAnswer(invocation -> {
 			callback.set(invocation.getArgument(0));
 			return null;
@@ -266,12 +269,14 @@ public void testInboundOptionsApplied() throws Exception {
 				"baz", "fix");
 		QueueChannel outputChannel = new QueueChannel();
 		adapter.setOutputChannel(outputChannel);
+		QueueChannel errorChannel = new QueueChannel();
+		adapter.setErrorChannel(errorChannel);
 		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
 		taskScheduler.initialize();
 		adapter.setTaskScheduler(taskScheduler);
 		adapter.setBeanFactory(mock(BeanFactory.class));
 		ApplicationEventPublisher applicationEventPublisher = mock(ApplicationEventPublisher.class);
-		final BlockingQueue<MqttIntegrationEvent> events = new LinkedBlockingQueue<MqttIntegrationEvent>();
+		final BlockingQueue<MqttIntegrationEvent> events = new LinkedBlockingQueue<>();
 		willAnswer(invocation -> {
 			events.add(invocation.getArgument(0));
 			return null;
@@ -294,6 +299,33 @@ public void testInboundOptionsApplied() throws Exception {
 		assertThat(event).isInstanceOf(MqttSubscribedEvent.class);
 		assertThat(((MqttSubscribedEvent) event).getMessage()).isEqualTo("Connected and subscribed to [baz, fix]");
 
+		adapter.setConverter(new MqttMessageConverter() {
+
+			@Override public Message<?> toMessage(String topic, MqttMessage mqttMessage) {
+				return null;
+			}
+
+			@Override public Object fromMessage(Message<?> message, Class<?> targetClass) {
+				return null;
+			}
+
+			@Override public Message<?> toMessage(Object payload, MessageHeaders headers) {
+				return null;
+			}
+
+
+		});
+
+		callback.get().messageArrived("baz", message);
+
+		ErrorMessage errorMessage = (ErrorMessage) errorChannel.receive(0);
+		assertThat(errorMessage).isNotNull()
+				.extracting(ErrorMessage::getPayload)
+				.isInstanceOf(IllegalStateException.class);
+		IllegalStateException exception = (IllegalStateException) errorMessage.getPayload();
+		assertThat(exception).hasMessage("'MqttMessageConverter' returned 'null'");
+		assertThat(errorMessage.getOriginalMessage().getPayload()).isSameAs(message);
+
 		// lose connection and make first reconnect fail
 		failConnection.set(true);
 		RuntimeException e = new RuntimeException("foo");
@@ -416,7 +448,7 @@ public void testReconnect() throws Exception {
 		// the following assertion should be equalTo, but leq to protect against a slow CI server
 		assertThat(attemptingReconnectCount.get()).isLessThanOrEqualTo(2);
 		AtomicReference<Object> failed = new AtomicReference<>();
-		adapter.setApplicationEventPublisher(event -> failed.set(event));
+		adapter.setApplicationEventPublisher(failed::set);
 		adapter.connectionLost(new IllegalStateException());
 		assertThat(failed.get()).isInstanceOf(MqttConnectionFailedEvent.class);
 		adapter.stop();
@@ -448,12 +480,11 @@ public void testSubscribeFailure() throws Exception {
 		new DirectFieldAccessor(client).setPropertyValue("aClient", aClient);
 		willAnswer(new CallsRealMethods()).given(client).connect(any(MqttConnectOptions.class));
 		willAnswer(new CallsRealMethods()).given(client).subscribe(any(String[].class), any(int[].class));
-		willAnswer(new CallsRealMethods()).given(client).subscribe(any(String[].class), any(int[].class),
-				(IMqttMessageListener[]) isNull());
+		willAnswer(new CallsRealMethods()).given(client).subscribe(any(String[].class), any(int[].class), isNull());
 		willReturn(alwaysComplete).given(aClient).connect(any(MqttConnectOptions.class), any(), any());
 
 		IMqttToken token = mock(IMqttToken.class);
-		given(token.getGrantedQos()).willReturn(new int[] { 0x80 });
+		given(token.getGrantedQos()).willReturn(new int[]{ 0x80 });
 		willReturn(token).given(aClient).subscribe(any(String[].class), any(int[].class), isNull(), isNull(), any());
 
 		MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory,
@@ -465,11 +496,11 @@ public void testSubscribeFailure() throws Exception {
 		}, m -> m.getName().equals("connectAndSubscribe"));
 		assertThat(method.get()).isNotNull();
 		Condition<InvocationTargetException> subscribeFailed = new Condition<>(ex ->
-			((MqttException) ex.getCause()).getReasonCode() == MqttException.REASON_CODE_SUBSCRIBE_FAILED,
-			"expected the reason code to be REASON_CODE_SUBSCRIBE_FAILED");
+				((MqttException) ex.getCause()).getReasonCode() == MqttException.REASON_CODE_SUBSCRIBE_FAILED,
+				"expected the reason code to be REASON_CODE_SUBSCRIBE_FAILED");
 		assertThatExceptionOfType(InvocationTargetException.class).isThrownBy(() -> method.get().invoke(adapter))
-			.withCauseInstanceOf(MqttException.class)
-			.is(subscribeFailed);
+				.withCauseInstanceOf(MqttException.class)
+				.is(subscribeFailed);
 	}
 
 	@Test
@@ -502,7 +533,7 @@ public void testDifferentQos() throws Exception {
 		willReturn(alwaysComplete).given(aClient).connect(any(MqttConnectOptions.class), any(), any());
 
 		IMqttToken token = mock(IMqttToken.class);
-		given(token.getGrantedQos()).willReturn(new int[] { 2, 0 });
+		given(token.getGrantedQos()).willReturn(new int[]{ 2, 0 });
 		willReturn(token).given(aClient).subscribe(any(String[].class), any(int[].class), isNull(), isNull(), any());
 
 		MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory,
@@ -537,7 +568,7 @@ public IMqttClient getClientInstance(String uri, String clientId) throws MqttExc
 
 		};
 		MqttConnectOptions connectOptions = new MqttConnectOptions();
-		connectOptions.setServerURIs(new String[] { "tcp://localhost:1883" });
+		connectOptions.setServerURIs(new String[]{ "tcp://localhost:1883" });
 		if (cleanSession != null) {
 			connectOptions.setCleanSession(cleanSession);
 		}
@@ -554,7 +585,7 @@ public IMqttClient getClientInstance(String uri, String clientId) throws MqttExc
 		return adapter;
 	}
 
-	private MqttPahoMessageHandler buildAdapterOut(final IMqttAsyncClient client) throws MqttException {
+	private MqttPahoMessageHandler buildAdapterOut(final IMqttAsyncClient client) {
 		DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory() {
 
 			@Override
@@ -564,7 +595,7 @@ public IMqttAsyncClient getAsyncClientInstance(String uri, String clientId) thro
 
 		};
 		MqttConnectOptions connectOptions = new MqttConnectOptions();
-		connectOptions.setServerURIs(new String[] { "tcp://localhost:1883" });
+		connectOptions.setServerURIs(new String[]{ "tcp://localhost:1883" });
 		factory.setConnectionOptions(connectOptions);
 		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("client", factory);
 		adapter.setDefaultTopic("foo");

src/reference/asciidoc/mqtt.adoc

@@ -145,6 +145,9 @@ A new application context reverts to the configured settings.
 
 Changing the topics while the adapter is stopped (or disconnected from the broker) takes effect the next time a connection is established.
 
+Starting with version `5.2.11`, when the message converter throws an exception or returns `null` from the `MqttMessage` conversion, the `MqttPahoMessageDrivenChannelAdapter` sends an `ErrorMessage` into the `errorChannel`, if provided.
+Re-throws this conversion error otherwise into an MQTT client callback.
+
 ==== Configuring with Java Configuration
 
 The following Spring Boot application shows an example of how to configure the inbound adapter with Java configuration: