You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by tu...@apache.org on 2021/01/06 15:47:50 UTC
[nifi] branch main updated: NIFI-7894 - ConsumeMQTT - allow EL on
Client ID property with shared subscription
This is an automated email from the ASF dual-hosted git repository.
turcsanyi 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 80eb570 NIFI-7894 - ConsumeMQTT - allow EL on Client ID property with shared subscription
80eb570 is described below
commit 80eb570ee10a0ee40e154699a712f5e414d38c44
Author: Pierre Villard <pi...@gmail.com>
AuthorDate: Thu Oct 8 11:32:06 2020 +0200
NIFI-7894 - ConsumeMQTT - allow EL on Client ID property with shared subscription
This closes #4586.
Signed-off-by: Peter Turcsanyi <tu...@apache.org>
---
.../org/apache/nifi/processors/mqtt/ConsumeMQTT.java | 12 ++++++++++--
.../mqtt/common/TestConsumeMqttCommon.java | 20 ++++++++++++++++++++
2 files changed, 30 insertions(+), 2 deletions(-)
diff --git a/nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/ConsumeMQTT.java b/nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/ConsumeMQTT.java
index ded89fe..5f93e65 100644
--- a/nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/ConsumeMQTT.java
+++ b/nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/ConsumeMQTT.java
@@ -198,9 +198,15 @@ public class ConsumeMQTT extends AbstractMQTTProcessor implements MqttCallback
}
final boolean clientIDSet = context.getProperty(PROP_CLIENTID).isSet();
+ final boolean clientIDwithEL = context.getProperty(PROP_CLIENTID).isExpressionLanguagePresent();
final boolean groupIDSet = context.getProperty(PROP_GROUPID).isSet();
- if (clientIDSet && groupIDSet) {
- results.add(new ValidationResult.Builder().subject("Client ID and Group ID").valid(false).explanation("if client ID is not unique, multiple nodes cannot join the consumer group").build());
+ if (!clientIDwithEL && clientIDSet && groupIDSet) {
+ results.add(new ValidationResult.Builder()
+ .subject("Client ID and Group ID").valid(false)
+ .explanation("if client ID is not unique, multiple nodes cannot join the consumer group (if you want "
+ + "to set the client ID, please use expression language to make sure each node in the NiFi "
+ + "cluster gets a unique client ID with something like ${hostname()}).")
+ .build());
}
return results;
@@ -230,6 +236,8 @@ public class ConsumeMQTT extends AbstractMQTTProcessor implements MqttCallback
if (context.getProperty(PROP_GROUPID).isSet()) {
topicPrefix = "$share/" + context.getProperty(PROP_GROUPID).getValue() + "/";
+ } else {
+ topicPrefix = "";
}
scheduled.set(true);
diff --git a/nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/common/TestConsumeMqttCommon.java b/nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/common/TestConsumeMqttCommon.java
index 71b6814..5518193 100644
--- a/nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/common/TestConsumeMqttCommon.java
+++ b/nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/common/TestConsumeMqttCommon.java
@@ -27,6 +27,7 @@ import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.provenance.ProvenanceEventType;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
import org.eclipse.paho.client.mqttv3.IMqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.junit.Test;
@@ -60,6 +61,25 @@ public abstract class TestConsumeMqttCommon {
public abstract void internalPublish(PublishMessage publishMessage);
@Test
+ public void testClientIDConfiguration() {
+ TestRunner runner = TestRunners.newTestRunner(ConsumeMQTT.class);
+ runner.setProperty(ConsumeMQTT.PROP_BROKER_URI, "tcp://localhost:1883");
+ runner.setProperty(ConsumeMQTT.PROP_CLIENTID, "TestClient");
+ runner.setProperty(ConsumeMQTT.PROP_TOPIC_FILTER, "testTopic");
+ runner.setProperty(ConsumeMQTT.PROP_MAX_QUEUE_SIZE, "100");
+ runner.assertValid();
+
+ runner.setProperty(ConsumeMQTT.PROP_GROUPID, "group");
+ runner.assertNotValid();
+
+ runner.setProperty(ConsumeMQTT.PROP_CLIENTID, "${hostname()}");
+ runner.assertValid();
+
+ runner.removeProperty(ConsumeMQTT.PROP_CLIENTID);
+ runner.assertValid();
+ }
+
+ @Test
public void testLastWillConfig() throws Exception {
testRunner.setProperty(ConsumeMQTT.PROP_LAST_WILL_MESSAGE, "lastWill message");
testRunner.assertNotValid();