[nifi] branch main updated: NIFI-7894 - ConsumeMQTT - allow EL on Client ID property with shared subscription

[turcsanyi]

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 <pierre.villard...@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 <turcsa...@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();