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

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.
@@ -40,6 +40,8 @@
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 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;
 
 /**
@@ -182,7 +184,7 @@ protected void doStart() {
 		}
 		catch (Exception e) {
 			logger.error("Exception while connecting and subscribing, retrying", e);
-			this.scheduleReconnect();
+			scheduleReconnect();
 		}
 	}
 
@@ -249,7 +251,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();
@@ -378,19 +380,52 @@ public synchronized void connectionLost(Throwable cause) {
 
 	@Override
 	public void messageArrived(String topic, MqttMessage mqttMessage) {
-		AbstractIntegrationMessageBuilder<?> builder = getConverter().toMessageBuilder(topic, mqttMessage);
-		if (this.manualAcks) {
-			builder.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK,
-					new AcknowledgmentImpl(mqttMessage.getId(), mqttMessage.getQos(), this.client));
+		AbstractIntegrationMessageBuilder<?> builder = toMessageBuilder(topic, mqttMessage);
+		if (builder != null) {
+			if (this.manualAcks) {
+				builder.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK,
+						new AcknowledgmentImpl(mqttMessage.getId(), mqttMessage.getQos(), this.client));
+			}
+			Message<?> message = builder.build();
+			try {
+				sendMessage(message);
+			}
+			catch (RuntimeException ex) {
+				logger.error("Unhandled exception for " + message.toString(), ex);
+				throw ex;
+			}
 		}
-		Message<?> message = builder.build();
+	}
+
+	private AbstractIntegrationMessageBuilder<?> toMessageBuilder(String topic, MqttMessage mqttMessage) {
+		AbstractIntegrationMessageBuilder<?> builder = null;
+		Exception conversionError = null;
 		try {
-			sendMessage(message);
+			builder = getConverter().toMessageBuilder(topic, mqttMessage);
 		}
-		catch (RuntimeException e) {
-			logger.error("Unhandled exception for " + message.toString(), e);
-			throw e;
+		catch (Exception ex) {
+			conversionError = ex;
+		}
+
+		if (builder == null && conversionError == null) {
+			conversionError = new IllegalStateException("'MqttMessageConverter' returned 'null'");
+		}
+
+		if (conversionError != null) {
+			GenericMessage<MqttMessage> message = new GenericMessage<>(mqttMessage);
+			if (!sendErrorMessageIfNecessary(message, conversionError)) {
+				MessageConversionException conversionException;
+				if (conversionError instanceof MessageConversionException) {
+					conversionException = (MessageConversionException) conversionError;
+				}
+				else {
+					conversionException = new MessageConversionException(message, "Failed to convert from MQTT Message",
+							conversionError);
+				}
+				throw conversionException;
+			}
 		}
+		return builder;
 	}
 
 	@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.
@@ -87,9 +87,13 @@
 import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
 import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
 import org.springframework.integration.mqtt.support.MqttHeaderAccessor;
+import org.springframework.integration.mqtt.support.MqttMessageConverter;
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 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;
@@ -104,7 +108,7 @@
  */
 public class MqttAdapterTests {
 
-	private IMqttToken alwaysComplete;
+	private final IMqttToken alwaysComplete;
 
 	{
 		ProxyFactoryBean pfb = new ProxyFactoryBean();
@@ -196,12 +200,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();
@@ -256,7 +260,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;
@@ -269,12 +273,14 @@ public void testInboundOptionsApplied() throws Exception {
 		adapter.setManualAcks(true);
 		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;
@@ -302,6 +308,39 @@ 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 AbstractIntegrationMessageBuilder<?> toMessageBuilder(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(Message::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");
@@ -424,7 +463,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();
@@ -456,8 +495,7 @@ 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);
@@ -562,7 +600,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

src/reference/asciidoc/mqtt.adoc

@@ -154,7 +154,7 @@ Starting with version 5.3, you can set the `manualAcks` property to true.
 Often used to asynchronously acknowledge delivery.
 When set to `true`, header (`IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK`) is added to the message with the value being a `SimpleAcknowledgment`.
 You must invoke the `acknowledge()` method to complete the delivery.
-See the Javadocs for `IMqppClient` `setManualAcks()` and `messageArrivedComplete()` for more information.
+See the Javadocs for `IMqttClient` `setManualAcks()` and `messageArrivedComplete()` for more information.
 For convenience a header accessor is provided:
 
 ====
@@ -164,6 +164,9 @@ StaticMessageHeaderAccessor.acknowledgment(someMessage).acknowledge();
 ----
 ====
 
+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: