GH-3181: MQTT: Support MANUAL Acks

Resolves https://github.com/spring-projects/spring-integration/issues/3181

* Doc polishing

* Rework acknowledgment into the existing `AcknowledgmentCallback`.

* Fix javadocs and doc linFix javadocs and doc linkk

* Doc polishing; explain uses of `ACKNOWLEDGMENT_CALLBACK` header.

spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java

@@ -65,6 +65,11 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
 
 	public static final String DELIVERY_ATTEMPT = "deliveryAttempt";
 
+	/**
+	 * A callback to acknowledge message delivery. The type of the header value depends on
+	 * the context in which the header is used. See the reference manual for more
+	 * information.
+	 */
 	public static final String ACKNOWLEDGMENT_CALLBACK = "acknowledgmentCallback";
 
 	/**

spring-integration-core/src/main/java/org/springframework/integration/StaticMessageHeaderAccessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2020 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.
@@ -21,6 +21,7 @@ import java.util.UUID;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.springframework.integration.acks.AcknowledgmentCallback;
+import org.springframework.integration.acks.SimpleAcknowledgment;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
@@ -107,6 +108,12 @@ public final class StaticMessageHeaderAccessor {
 				AcknowledgmentCallback.class);
 	}
 
+	@Nullable
+	public static SimpleAcknowledgment getAcknowledgment(Message<?> message) {
+		return message.getHeaders().get(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK,
+				SimpleAcknowledgment.class);
+	}
+
 	@SuppressWarnings("unchecked")
 	@Nullable
 	public static <T> T getSourceData(Message<?> message) {

spring-integration-core/src/main/java/org/springframework/integration/acks/AcknowledgmentCallback.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 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.
@@ -25,7 +25,7 @@ package org.springframework.integration.acks;
  *
  */
 @FunctionalInterface
-public interface AcknowledgmentCallback {
+public interface AcknowledgmentCallback extends SimpleAcknowledgment {
 
 	/**
 	 * Acknowledge the message.
@@ -33,6 +33,11 @@ public interface AcknowledgmentCallback {
 	 */
 	void acknowledge(Status status);
 
+	@Override
+	default void acknowledge() {
+		acknowledge(Status.ACCEPT);
+	}
+
 	/**
 	 * Implementations must implement this to indicate when the ack has been
 	 * processed by the user so that the framework can auto-ack if needed.

spring-integration-core/src/main/java/org/springframework/integration/acks/SimpleAcknowledgment.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2020 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.acks;
+
+/**
+ * Opaque object for manually acknowledging.
+ *
+ * @author Gary Russell
+ * @since 5.3
+ *
+ */
+@FunctionalInterface
+public interface SimpleAcknowledgment {
+
+	/**
+	 * Acknowledge the message delivery.
+	 */
+	void acknowledge();
+
+}

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -48,6 +48,7 @@ public class MqttMessageDrivenChannelAdapterParser extends AbstractChannelAdapte
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "qos");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "recovery-interval");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "manual-acks");
 
 		return builder.getBeanDefinition();
 	}

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

@@ -30,11 +30,14 @@ import org.eclipse.paho.client.mqttv3.MqttMessage;
 
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.acks.SimpleAcknowledgment;
 import org.springframework.integration.mqtt.core.ConsumerStopAction;
 import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
 import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
 import org.springframework.integration.mqtt.event.MqttConnectionFailedEvent;
 import org.springframework.integration.mqtt.event.MqttSubscribedEvent;
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
@@ -71,6 +74,8 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 
 	private long disconnectCompletionTimeout = DISCONNECT_COMPLETION_TIMEOUT;
 
+	private boolean manualAcks;
+
 	private volatile IMqttClient client;
 
 	private volatile ScheduledFuture<?> reconnectFuture;
@@ -152,6 +157,15 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 		this.recoveryInterval = recoveryInterval;
 	}
 
+	/**
+	 * Set the acknowledgment mode to manual.
+	 * @param manualAcks true for manual acks.
+	 * @since 5.3
+	 */
+	public void setManualAcks(boolean manualAcks) {
+		this.manualAcks = manualAcks;
+	}
+
 	/**
 	 * @since 4.2.2
 	 */
@@ -263,6 +277,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 		String[] topics = getTopic();
 		try {
 			this.client.connect(connectionOptions);
+			this.client.setManualAcks(this.manualAcks);
 			int[] requestedQos = getQos();
 			int[] grantedQos = Arrays.copyOf(requestedQos, requestedQos.length);
 			this.client.subscribe(topics, grantedQos);
@@ -365,7 +380,12 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 
 	@Override
 	public void messageArrived(String topic, MqttMessage mqttMessage) {
-		Message<?> message = this.getConverter().toMessage(topic, mqttMessage);
+		AbstractIntegrationMessageBuilder<?> builder = getConverter().toMessageBuilder(topic, mqttMessage);
+		if (this.manualAcks) {
+			builder.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK,
+					new AcknowledgmentImpl(mqttMessage.getId(), mqttMessage.getQos(), this.client));
+		}
+		Message<?> message = builder.build();
 		try {
 			sendMessage(message);
 		}
@@ -379,4 +399,46 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 	public void deliveryComplete(IMqttDeliveryToken token) {
 	}
 
+	/**
+	 * Used to complete message arrival when {@link AckMode#MANUAL}.
+	 *
+	 * @since 5.3
+	 */
+	private static class AcknowledgmentImpl implements SimpleAcknowledgment {
+
+		private final int id;
+
+		private final int qos;
+
+		private final IMqttClient ackClient;
+
+		/**
+		 * Construct an instance with the provided properties.
+		 * @param id the message id.
+		 * @param qos the message QOS.
+		 * @param client the client.
+		 */
+		AcknowledgmentImpl(int id, int qos, IMqttClient client) {
+			this.id = id;
+			this.qos = qos;
+			this.ackClient = client;
+		}
+
+		@Override
+		public void acknowledge() {
+			if (this.ackClient != null) {
+				try {
+					this.ackClient.messageArrivedComplete(this.id, this.qos);
+				}
+				catch (MqttException e) {
+					throw new IllegalStateException(e);
+				}
+			}
+			else {
+				throw new IllegalStateException("Client has changed");
+			}
+		}
+
+	}
+
 }

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -206,6 +206,11 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa
 
 	@Override
 	public Message<?> toMessage(String topic, MqttMessage mqttMessage) {
+		return toMessageBuilder(topic, mqttMessage).build();
+	}
+
+	@Override
+	public AbstractIntegrationMessageBuilder<?> toMessageBuilder(String topic, MqttMessage mqttMessage) {
 		try {
 			AbstractIntegrationMessageBuilder<?> messageBuilder;
 			if (this.bytesMessageMapper != null) {
@@ -219,13 +224,14 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa
 								.withPayload(mqttBytesToPayload(mqttMessage));
 			}
 			messageBuilder
+					.setHeader(MqttHeaders.ID, mqttMessage.getId())
 					.setHeader(MqttHeaders.RECEIVED_QOS, mqttMessage.getQos())
 					.setHeader(MqttHeaders.DUPLICATE, mqttMessage.isDuplicate())
 					.setHeader(MqttHeaders.RECEIVED_RETAINED, mqttMessage.isRetained());
 			if (topic != null) {
 				messageBuilder.setHeader(MqttHeaders.RECEIVED_TOPIC, topic);
 			}
-			return messageBuilder.build();
+			return messageBuilder;
 		}
 		catch (Exception e) {
 			throw new MessageConversionException("failed to convert object to Message", e);

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttHeaderAccessor.java

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2020 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.mqtt.support;
+
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+
+/**
+ * Helper for typed access to incoming MQTT message headers.
+ *
+ * @author Gary Russell
+ * @since 5.3
+ *
+ */
+public final class MqttHeaderAccessor {
+
+	private MqttHeaderAccessor() {
+	}
+
+	/**
+	 * Return the received topic header.
+	 * @param message the message.
+	 * @return the header.
+	 */
+	@Nullable
+	public static String receivedTopic(Message<?> message) {
+		return message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC, String.class);
+	}
+
+	/**
+	 * Return the MQTT message id.
+	 * @param message the message.
+	 * @return the header.
+	 */
+	@Nullable
+	public static Integer id(Message<?> message) {
+		return message.getHeaders().get(MqttHeaders.ID, Integer.class);
+	}
+
+	/**
+	 * Return the received QOS header.
+	 * @param message the message.
+	 * @return the header.
+	 */
+	@Nullable
+	public static Integer receivedQos(Message<?> message) {
+		return message.getHeaders().get(MqttHeaders.RECEIVED_QOS, Integer.class);
+	}
+
+	/**
+	 * Return the received retained header.
+	 * @param message the message.
+	 * @return the header.
+	 */
+	@Nullable
+	public static Boolean receivedRetained(Message<?> message) {
+		return message.getHeaders().get(MqttHeaders.RECEIVED_RETAINED, Boolean.class);
+	}
+
+	/**
+	 * Return the duplicate header.
+	 * @param message the message.
+	 * @return the header.
+	 */
+	@Nullable
+	public static Boolean duplicate(Message<?> message) {
+		return message.getHeaders().get(MqttHeaders.DUPLICATE, Boolean.class);
+	}
+
+}

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttHeaders.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -30,6 +30,8 @@ public final class MqttHeaders {
 
 	public static final String QOS = PREFIX + "qos";
 
+	public static final String ID = PREFIX + "id";
+
 	public static final String RECEIVED_QOS = PREFIX + "receivedQos";
 
 	public static final String DUPLICATE = PREFIX + "duplicate";

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttMessageConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -19,6 +19,7 @@ package org.springframework.integration.mqtt.support;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
 
 import org.springframework.integration.handler.MessageProcessor;
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.converter.MessageConverter;
 
@@ -35,12 +36,20 @@ public interface MqttMessageConverter extends MessageConverter {
 	/**
 	 * Convert to a Message.
 	 *
-	 * @param topic The topic.
-	 * @param mqttMessage The MQTT message.
-	 * @return The Message.
+	 * @param topic the topic.
+	 * @param mqttMessage the MQTT message.
+	 * @return the Message.
 	 */
 	Message<?> toMessage(String topic, MqttMessage mqttMessage);
 
+	/**
+	 * Convert to a message builder.
+	 * @param topic the topic.
+	 * @param mqttMessage the MQTT message.
+	 * @return the builder.
+	 */
+	AbstractIntegrationMessageBuilder<?> toMessageBuilder(String topic, MqttMessage mqttMessage);
+
 	static MessageProcessor<Integer> defaultQosProcessor() {
 		return message -> message.getHeaders().get(MqttHeaders.QOS, Integer.class);
 	}

spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/spring-integration-mqtt.xsd

@@ -89,6 +89,14 @@
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
+			<xsd:attribute name="manual-acks" type="xsd:boolean" default="false">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+						Acknowledgment mode (true for manual acks).
+					]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
 		</xsd:complexType>
 	</xsd:element>
 

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

@@ -74,6 +74,7 @@ import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.StaticMessageHeaderAccessor;
 import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.handler.MessageProcessor;
@@ -85,6 +86,7 @@ import org.springframework.integration.mqtt.event.MqttSubscribedEvent;
 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.MqttHeaderAccessor;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandlingException;
@@ -264,6 +266,7 @@ public class MqttAdapterTests {
 
 		MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory,
 				"baz", "fix");
+		adapter.setManualAcks(true);
 		QueueChannel outputChannel = new QueueChannel();
 		adapter.setOutputChannel(outputChannel);
 		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
@@ -290,6 +293,11 @@ public class MqttAdapterTests {
 		assertThat(outMessage).isNotNull();
 		assertThat(outMessage.getPayload()).isEqualTo("qux");
 
+		StaticMessageHeaderAccessor.getAcknowledgment(outMessage).acknowledge();
+		verify(client).setManualAcks(true);
+		verify(client)
+				.messageArrivedComplete(MqttHeaderAccessor.id(outMessage), MqttHeaderAccessor.receivedQos(outMessage));
+
 		MqttIntegrationEvent event = events.poll(10, TimeUnit.SECONDS);
 		assertThat(event).isInstanceOf(MqttSubscribedEvent.class);
 		assertThat(((MqttSubscribedEvent) event).getMessage()).isEqualTo("Connected and subscribed to [baz, fix]");

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests-context.xml

@@ -12,6 +12,7 @@
 
 	<int-mqtt:message-driven-channel-adapter id="noTopicsAdapter"
 		auto-startup="false"
+		manual-acks="true"
 		client-id="foo"
 		url="tcp://localhost:1883"
 		client-factory="clientFactory"

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -79,6 +79,7 @@ public class MqttMessageDrivenChannelAdapterParserTests {
 		assertThat(TestUtils.getPropertyValue(noTopicsAdapter, "outputChannel")).isSameAs(out);
 		assertThat(TestUtils.getPropertyValue(noTopicsAdapter, "clientFactory")).isSameAs(clientFactory);
 		assertThat(TestUtils.getPropertyValue(this.noTopicsAdapter, "recoveryInterval")).isEqualTo(5000);
+		assertThat(TestUtils.getPropertyValue(this.noTopicsAdapter, "manualAcks", Boolean.class)).isTrue();
 	}
 
 	@Test
@@ -89,6 +90,7 @@ public class MqttMessageDrivenChannelAdapterParserTests {
 		assertThat(TestUtils.getPropertyValue(noTopicsAdapterDefaultCF, "topics", Collection.class).size())
 				.isEqualTo(0);
 		assertThat(TestUtils.getPropertyValue(noTopicsAdapterDefaultCF, "outputChannel")).isSameAs(out);
+		assertThat(TestUtils.getPropertyValue(this.noTopicsAdapterDefaultCF, "manualAcks", Boolean.class)).isFalse();
 	}
 
 	@Test

src/reference/asciidoc/message.adoc

@@ -171,8 +171,8 @@ ACKNOWLEDGMENT_CALLBACK
 | o.s.i.support.
 Acknowledgment
 Callback
-| If a message source supports it, a call back to accept, reject, or requeue a message.
-See <<./polling-consumer.adoc#deferred-acks-message-source,Deferred Acknowledgment Pollable Message Source>>.
+| If an inbound endpoint supports it, a call back to accept, reject, or requeue a message.
+See <<./polling-consumer.adoc#deferred-acks-message-source,Deferred Acknowledgment Pollable Message Source>> and <<./mqtt.adoc#mqtt-ack-mode,MQTT Manual Acks>>.
 |===
 
 Convenient typed getters for some of these headers are provided on the `IntegrationMessageHeaderAccessor` class, as the following example shows:

src/reference/asciidoc/mqtt.adoc

@@ -74,6 +74,7 @@ The following listing shows the available attributes:
     send-timeout="123"  <7>
     error-channel="errors"  <8>
     recovery-interval="10000"  <9>
+    manual-acks="false" <10>
     channel="out" />
 ----
 
@@ -97,6 +98,7 @@ The payload is a `MessagingException` that contains the failed message and cause
 <9> The recovery interval.
 It controls the interval at which the adapter attempts to reconnect after a failure.
 It defaults to `10000ms` (ten seconds).
+<10> The acknowledgment mode; set to true for manual acknowledgment.
 ====
 
 NOTE: Starting with version 4.1, you can omit the URL.
@@ -145,6 +147,23 @@ 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.
 
+[[mqtt-ack-mode]]
+==== Manual Acks
+
+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.
+For convenience a header accessor is provided:
+
+====
+[source, java]
+----
+StaticMessageHeaderAccessor.acknowledgment(someMessage).acknowledge();
+----
+====
+
 ==== Configuring with Java Configuration
 
 The following Spring Boot application shows an example of how to configure the inbound adapter with Java configuration:
@@ -278,7 +297,7 @@ The default is `headers['mqtt_topic']`.
 <11> When `true`, the caller does not block.
 Rather, it waits for delivery confirmation when a message is sent.
 The default is `false` (the send blocks until delivery is confirmed).
-<12> When `async` and `async-events` are both `true`, an `MqttMessageSentEvent` is emitted (See <<events>>).
+<12> When `async` and `async-events` are both `true`, an `MqttMessageSentEvent` is emitted (See <<mqtt-events>>).
 It contains the message, the topic, the `messageId` generated by the client library, the `clientId`, and the `clientInstance` (incremented each time the client is connected).
 When the delivery is confirmed by the client library, an `MqttMessageDeliveredEvent` is emitted.
 It contains the the `messageId`, the `clientId`, and the `clientInstance`, enabling delivery to be correlated with the send.
@@ -373,7 +392,7 @@ public class MqttJavaApplication {
 ----
 ====
 
-[[events]]
+[[mqtt-events]]
 === Events
 
 Certain application events are published by the adapters.
@@ -381,5 +400,6 @@ Certain application events are published by the adapters.
 * `MqttConnectionFailedEvent` - published by both adapters if we fail to connect or a connection is subsequently lost.
 * `MqttMessageSentEvent` - published by the outbound adapter when a message has been sent, if running in asynchronous mode.
 * `MqttMessageDeliveredEvent` - published by the outbound adapter when the client indicates that a message has been delivered, if running in asynchronous mode.
+* `MqttSubscribedEvent` - published by the inbound adapter after subscribing to the topics.
 
 These events can be received by an `ApplicationListener<MqttIntegrationEvent>` or with an `@EventListener` method.

src/reference/asciidoc/polling-consumer.adoc

@@ -60,7 +60,7 @@ Starting with version 5.0.1, certain modules provide `MessageSource` implementat
 This is currently limited to the `AmqpMessageSource` and the `KafkaMessageSource` provided by the `spring-integration-kafka` https://github.com/spring-projects/spring-integration-kafka[extension project].
 
 With these message sources, the `IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK` header (see <<./message.adoc#message-header-accessor,`MessageHeaderAccessor` API>>) is added to the message.
-The value of the header is an instance of `AcknowledgmentCallback`, as the following example shows:
+When used with pollable message sources, the value of the header is an instance of `AcknowledgmentCallback`, as the following example shows:
 
 [source, java]
 ----

src/reference/asciidoc/whats-new.adoc

@@ -116,3 +116,13 @@ See <<./rsocket.adoc#rsocket-inbound,RSocket Inbound Gateway>> for more informat
 
 A `LeaderInitiatorFactoryBean` (as well as its XML `<int-zk:leader-listener>`) exposes a `candidate` option for more control over a `Candidate` configuration.
 See <<./zookeeper.adoc#zk-leadership,Leadership event handling>> for more information.
+
+[[x5.3-mqtt]]
+=== MQTT Changes
+
+The inbound channel adapter can now be configured to provide user control over when a message is acknowledged as being delivered.
+See <<./mqtt.adoc#mqtt-ack-mode,Manual Acks>> for more information.
+
+The outbound adapter now publishes a `MqttConnectionFailedEvent` when a connection can't be created, or is lost.
+Previously, only the inbound adapter did so.
+See <<./mqtt.adoc#mqtt-events,MQTT Events>>.