This is an automated email from the ASF dual-hosted git repository.
pvillard31 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 56ec4ca59d3 NIFI-16421 Fixed ConsumeMQTT loses the in-flight message
on reconnect… (#11780)
56ec4ca59d3 is described below
commit 56ec4ca59d3710faf3f4159b3b3a1761cafb1b86
Author: Peter Turcsanyi <[email protected]>
AuthorDate: Fri Oct 9 11:54:09 2026 +0200
NIFI-16421 Fixed ConsumeMQTT loses the in-flight message on reconnect…
(#11780)
* NIFI-16421 Fixed ConsumeMQTT loses the in-flight message on reconnect
with Resume Session
---
.../apache/nifi/processors/mqtt/ConsumeMQTT.java | 3 ++-
.../mqtt/adapters/HiveMqV5ClientAdapter.java | 25 ++++++++++++----------
.../mqtt/adapters/PahoMqttClientAdapter.java | 11 +++++++---
.../nifi/processors/mqtt/common/MqttClient.java | 11 ++++++++--
.../nifi/processors/mqtt/TestConsumeMQTT.java | 19 ++++++++++++++++
.../processors/mqtt/common/MqttTestClient.java | 6 +++++-
6 files changed, 57 insertions(+), 18 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/ConsumeMQTT.java
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/ConsumeMQTT.java
index 7763eb8a4c9..4faa1d005b2 100644
---
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/ConsumeMQTT.java
+++
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/ConsumeMQTT.java
@@ -388,8 +388,9 @@ public class ConsumeMQTT extends AbstractMQTTProcessor {
// non-null but not connected, so we need to handle each case and only
create a new client when it is null
try {
mqttClient = createMqttClient();
+ mqttClient.setReceivedMessageHandler(this::handleReceivedMessage);
mqttClient.connect();
- mqttClient.subscribe(topicPrefix + topicFilter, qos,
this::handleReceivedMessage);
+ mqttClient.subscribe(topicPrefix + topicFilter, qos);
} catch (Exception e) {
logger.error("Connection failed to {}. Yielding processor",
clientProperties.getRawBrokerUris(), e);
mqttClient = null; // prevent stuck processor when subscribe fails
diff --git
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/HiveMqV5ClientAdapter.java
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/HiveMqV5ClientAdapter.java
index 8894bbbef49..fc892cda5a8 100644
---
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/HiveMqV5ClientAdapter.java
+++
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/HiveMqV5ClientAdapter.java
@@ -16,6 +16,7 @@
*/
package org.apache.nifi.processors.mqtt.adapters;
+import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
import com.hivemq.client.mqtt.mqtt5.Mqtt5Client;
@@ -128,24 +129,14 @@ public class HiveMqV5ClientAdapter implements MqttClient {
}
@Override
- public void subscribe(String topicFilter, int qos,
ReceivedMqttMessageHandler handler) {
+ public void subscribe(String topicFilter, int qos) {
logger.debug("Subscribing to {} with QoS: {}", topicFilter, qos);
CompletableFuture<Mqtt5SubAck> futureAck =
mqtt5BlockingClient.toAsync().subscribeWith()
.topicFilter(topicFilter)
.qos(Objects.requireNonNull(MqttQos.fromCode(qos)))
- .callback(mqtt5Publish -> {
- final ReceivedMqttMessage receivedMessage = new
ReceivedMqttMessage(
- mqtt5Publish.getPayloadAsBytes(),
- mqtt5Publish.getQos().getCode(),
- mqtt5Publish.isRetain(),
- mqtt5Publish.getTopic().toString());
- handler.handleReceivedMessage(receivedMessage);
- })
.send();
- // Setting "listener" callback is only possible with async client,
though sending subscribe message
- // should happen in a blocking way to make sure the processor is
blocked until ack is not arrived.
try {
final Mqtt5SubAck ack =
futureAck.get(clientProperties.getConnectionTimeout(), TimeUnit.SECONDS);
logger.debug("Received mqtt5 subscribe ack: {}", ack);
@@ -154,6 +145,18 @@ public class HiveMqV5ClientAdapter implements MqttClient {
}
}
+ @Override
+ public void setReceivedMessageHandler(ReceivedMqttMessageHandler handler) {
+ mqtt5BlockingClient.toAsync().publishes(MqttGlobalPublishFilter.ALL,
mqtt5Publish -> {
+ final ReceivedMqttMessage receivedMessage = new
ReceivedMqttMessage(
+ mqtt5Publish.getPayloadAsBytes(),
+ mqtt5Publish.getQos().getCode(),
+ mqtt5Publish.isRetain(),
+ mqtt5Publish.getTopic().toString());
+ handler.handleReceivedMessage(receivedMessage);
+ });
+ }
+
private static Mqtt5BlockingClient createClient(URI brokerUri,
MqttClientProperties clientProperties, ComponentLog logger) throws TlsException
{
logger.debug("Creating Mqtt v5 client");
diff --git
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PahoMqttClientAdapter.java
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PahoMqttClientAdapter.java
index ddef128eb94..96cd82159e2 100644
---
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PahoMqttClientAdapter.java
+++
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PahoMqttClientAdapter.java
@@ -123,11 +123,9 @@ public class PahoMqttClientAdapter implements MqttClient {
}
@Override
- public void subscribe(String topicFilter, int qos,
ReceivedMqttMessageHandler handler) {
+ public void subscribe(String topicFilter, int qos) {
logger.debug("Subscribing to {} with QoS: {}", topicFilter, qos);
- client.setCallback(new ConsumerMqttCallback(handler));
-
try {
client.subscribe(topicFilter, qos);
} catch (org.eclipse.paho.client.mqttv3.MqttException e) {
@@ -135,6 +133,13 @@ public class PahoMqttClientAdapter implements MqttClient {
}
}
+ @Override
+ public void setReceivedMessageHandler(ReceivedMqttMessageHandler handler) {
+ logger.debug("Registering received message handler");
+
+ client.setCallback(new ConsumerMqttCallback(handler));
+ }
+
private static org.eclipse.paho.client.mqttv3.MqttClient createClient(URI
brokerUri, MqttClientProperties clientProperties, ComponentLog logger) {
logger.debug("Creating Mqtt v3 client");
diff --git
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/common/MqttClient.java
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/common/MqttClient.java
index 2b5c949531d..7c0a9376c26 100644
---
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/common/MqttClient.java
+++
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/common/MqttClient.java
@@ -57,7 +57,14 @@ public interface MqttClient {
* published at a lower quality of service will be received at
the published
* QoS. Messages published at a higher quality of service will
be received using
* the QoS specified on the subscribe.
- * @param handler that further processes the message received by the client
*/
- void subscribe(String topicFilter, int qos, ReceivedMqttMessageHandler
handler);
+ void subscribe(String topicFilter, int qos);
+
+ /**
+ * Sets the handler to process received MQTT messages.
+ * It must be set before the client is connected to the broker, otherwise
messages may be lost when a persistent session is used.
+ *
+ * @param handler the handler to process received MQTT messages
+ */
+ void setReceivedMessageHandler(ReceivedMqttMessageHandler handler);
}
diff --git
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/TestConsumeMQTT.java
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/TestConsumeMQTT.java
index ce3eed059e0..97d986b138b 100644
---
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/TestConsumeMQTT.java
+++
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/TestConsumeMQTT.java
@@ -647,6 +647,25 @@ public class TestConsumeMQTT {
assertTrue(mqttQueue.contains(mock));
}
+ @Test
+ void testConsumeMessageDeliveredOnConnect() throws Exception {
+ mqttTestClient = new
MqttTestClient(MqttTestClient.ConnectType.Subscriber) {
+ @Override
+ public void connect() {
+ super.connect();
+ publishMessage(STRING_MESSAGE, AT_LEAST_ONCE);
+ }
+ };
+ testRunner = initializeTestRunner(mqttTestClient);
+
+ testRunner.run();
+
+ final List<MockFlowFile> flowFiles =
testRunner.getFlowFilesForRelationship(ConsumeMQTT.REL_MESSAGE);
+ assertEquals(1, flowFiles.size());
+ final MockFlowFile flowFile = flowFiles.getFirst();
+ flowFile.assertContentEquals(STRING_MESSAGE);
+ }
+
@Test
void addTopicAttributesWithMultipleTopicSegments() {
final String topic = "home/livingroom/temperature";
diff --git
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/common/MqttTestClient.java
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/common/MqttTestClient.java
index 3941bedf8a8..c69e48906f2 100644
---
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/common/MqttTestClient.java
+++
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/common/MqttTestClient.java
@@ -73,9 +73,13 @@ public class MqttTestClient implements MqttClient {
}
@Override
- public void subscribe(String topicFilter, int qos,
ReceivedMqttMessageHandler handler) {
+ public void subscribe(String topicFilter, int qos) {
subscribedTopic = topicFilter;
subscribedQos = qos;
+ }
+
+ @Override
+ public void setReceivedMessageHandler(ReceivedMqttMessageHandler handler) {
receivedMqttMessageHandler = handler;
}
