Skip to content

NIFI-16203 Support multiple Topic Filters in ConsumeMQTT #11556

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
drewyh999 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

NIFI-16203 Support multiple Topic Filters in ConsumeMQTT #11556

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
82 changes: 77 additions & 5 deletions ...undle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/ConsumeMQTT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.processors.mqtt.common.AbstractMQTTProcessor;
import org.apache.nifi.processors.mqtt.common.MqttException;
import org.apache.nifi.processors.mqtt.common.MqttTopicSubscription;
import org.apache.nifi.processors.mqtt.common.ReceivedMqttMessage;
import org.apache.nifi.serialization.MalformedRecordException;
import org.apache.nifi.serialization.RecordReader;
Expand All @@ -67,6 +68,8 @@
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
Expand All @@ -75,6 +78,7 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import static org.apache.nifi.processors.mqtt.ConsumeMQTT.BROKER_ATTRIBUTE_KEY;
import static org.apache.nifi.processors.mqtt.ConsumeMQTT.IS_DUPLICATE_ATTRIBUTE_KEY;
Expand Down Expand Up @@ -133,7 +137,10 @@ public class ConsumeMQTT extends AbstractMQTTProcessor {

public static final PropertyDescriptor PROP_TOPIC_FILTER = new PropertyDescriptor.Builder()
.name("Topic Filter")
.description("The MQTT topic filter to designate the topics to subscribe to.")
.description("The MQTT topic filter to designate the topics to subscribe to. More than one can be supplied if comma separated, in which case a single SUBSCRIBE request "
+ "listing every filter is sent to the broker, avoiding the need for a separate processor and broker connection per topic. A value without a comma is used as a "
+ "single topic filter exactly as configured, while the entries of a comma separated value are trimmed. Because MQTT topic filters may legally contain a comma, "
+ "a Topic Filter containing one is interpreted as multiple filters; this is a rare edge case but should be kept in mind.")
.required(true)
.expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
.addValidator(StandardValidators.NON_BLANK_VALIDATOR)
Expand Down Expand Up @@ -192,6 +199,7 @@ public class ConsumeMQTT extends AbstractMQTTProcessor {
private volatile int qos;
private volatile String topicPrefix = "";
private volatile String topicFilter;
private volatile List<MqttTopicSubscription> topicSubscriptions = List.of();
private final AtomicBoolean scheduled = new AtomicBoolean(false);

private volatile BlockingQueue<ReceivedMqttMessage> mqttQueue;
Expand Down Expand Up @@ -288,6 +296,27 @@ public Collection<ValidationResult> customValidate(ValidationContext context) {
.build());
}

final String rawTopicFilter = context.getProperty(PROP_TOPIC_FILTER).evaluateAttributeExpressions().getValue();
if (rawTopicFilter != null) {
final List<String> topicFilters = parseTopicFilters(rawTopicFilter);
if (topicFilters.isEmpty()) {
results.add(new ValidationResult.Builder()
.subject(PROP_TOPIC_FILTER.getDisplayName())
.valid(false)
.explanation("at least one non-blank Topic Filter must be provided.")
.build());
} else {
final Set<String> uniqueTopicFilters = new HashSet<>(topicFilters);
if (uniqueTopicFilters.size() != topicFilters.size()) {
results.add(new ValidationResult.Builder()
.subject(PROP_TOPIC_FILTER.getDisplayName())
.valid(false)
.explanation("duplicate Topic Filters are not allowed: " + topicFilters)
.build());
}
}
}

return results;
}

Expand Down Expand Up @@ -319,6 +348,12 @@ public void onScheduled(final ProcessContext context) {
topicPrefix = "";
}

// The shared subscription prefix applies to an individual Topic Filter, so it has to be added to each of them
// separately rather than to the configured, potentially comma separated, value as a whole.
topicSubscriptions = new LinkedHashSet<>(parseTopicFilters(topicFilter)).stream()
.map(filter -> new MqttTopicSubscription(topicPrefix + filter, qos))
.collect(Collectors.toList());

scheduled.set(true);
}

Expand Down Expand Up @@ -389,14 +424,39 @@ private void initializeClient(ProcessContext context) {
try {
mqttClient = createMqttClient();
mqttClient.connect();
mqttClient.subscribe(topicPrefix + topicFilter, qos, this::handleReceivedMessage);
mqttClient.subscribe(topicSubscriptions, this::handleReceivedMessage);
} catch (Exception e) {
logger.error("Connection failed to {}. Yielding processor", clientProperties.getRawBrokerUris(), e);
mqttClient = null; // prevent stuck processor when subscribe fails
// A SUBSCRIBE carrying several Topic Filters can be granted partially, so the client may be connected and
// subscribed even though subscribe() failed. Disconnecting and closing it, rather than only dropping the
// reference, prevents an orphaned client from holding a broker connection and feeding the internal queue.
stopClient();
context.yield();
}
}

/**
* Splits the configured Topic Filter property, which may contain a comma-separated list of topic filters, into
* a list of non-blank topic filters, preserving duplicates so that {@link #customValidate(ValidationContext)} can
* flag them. A value without a comma is a single topic filter and is used verbatim, so existing configurations
* behave identically. Only the segments of a comma-separated value are trimmed, because leading and trailing
* whitespace is significant in an MQTT topic filter and trimming is merely a convenience for writing a list.
*/
private static List<String> parseTopicFilters(final String rawTopicFilters) {
if (rawTopicFilters.indexOf(',') < 0) {
return List.of(rawTopicFilters);
}

final List<String> topicFilters = new ArrayList<>();
for (final String topicFilter : rawTopicFilters.split(",", -1)) {
final String trimmedTopicFilter = topicFilter.trim();
if (!trimmedTopicFilter.isEmpty()) {
topicFilters.add(trimmedTopicFilter);
}
}
return topicFilters;
}

private void transferQueue(ProcessSession session) {
while (!mqttQueue.isEmpty()) {
final ReceivedMqttMessage mqttMessage = mqttQueue.peek();
Expand Down Expand Up @@ -430,7 +490,7 @@ private void transferQueueDemarcator(final ProcessContext context, final Process
}
});

session.getProvenanceReporter().receive(messageFlowfile, getTransitUri(topicPrefix, topicFilter));
session.getProvenanceReporter().receive(messageFlowfile, getTransitUri(getSubscribedTopicsForProvenance()));
session.transfer(messageFlowfile, REL_MESSAGE);
session.commitAsync();
}
Expand Down Expand Up @@ -607,7 +667,7 @@ private void transferQueueRecord(final ProcessContext context, final ProcessSess
}

session.putAllAttributes(flowFile, attributes);
session.getProvenanceReporter().receive(flowFile, getTransitUri(topicPrefix, topicFilter));
session.getProvenanceReporter().receive(flowFile, getTransitUri(getSubscribedTopicsForProvenance()));
session.transfer(flowFile, REL_MESSAGE);

final int count = recordCount.get();
Expand Down Expand Up @@ -645,6 +705,18 @@ private String getTransitUri(String... appends) {
return stringBuilder.toString();
}

/**
* Returns the subscribed topic filters, including the shared subscription prefix if any, as a single comma
* separated value. A FlowFile produced by the demarcator or record based code paths may aggregate messages of
* several topics, so the individual topic of a message cannot be used for those. For a single Topic Filter this
* returns exactly the previously reported value.
*/
private String getSubscribedTopicsForProvenance() {
return topicSubscriptions.stream()
.map(MqttTopicSubscription::topicFilter)
.collect(Collectors.joining(","));
}

private void handleReceivedMessage(ReceivedMqttMessage message) {
if (logger.isDebugEnabled()) {
byte[] payload = message.getPayload();
Expand Down
75 changes: 55 additions & 20 deletions ...cessors/src/main/java/org/apache/nifi/processors/mqtt/adapters/HiveMqV5ClientAdapter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,12 +22,16 @@
import com.hivemq.client.mqtt.mqtt5.Mqtt5ClientBuilder;
import com.hivemq.client.mqtt.mqtt5.message.connect.Mqtt5Connect;
import com.hivemq.client.mqtt.mqtt5.message.connect.Mqtt5ConnectBuilder;
import com.hivemq.client.mqtt.mqtt5.message.subscribe.Mqtt5Subscribe;
import com.hivemq.client.mqtt.mqtt5.message.subscribe.Mqtt5Subscription;
import com.hivemq.client.mqtt.mqtt5.message.subscribe.suback.Mqtt5SubAck;
import com.hivemq.client.mqtt.mqtt5.message.subscribe.suback.Mqtt5SubAckReasonCode;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processors.mqtt.common.MqttClient;
import org.apache.nifi.processors.mqtt.common.MqttClientProperties;
import org.apache.nifi.processors.mqtt.common.MqttException;
import org.apache.nifi.processors.mqtt.common.MqttProtocolScheme;
import org.apache.nifi.processors.mqtt.common.MqttTopicSubscription;
import org.apache.nifi.processors.mqtt.common.ReceivedMqttMessage;
import org.apache.nifi.processors.mqtt.common.ReceivedMqttMessageHandler;
import org.apache.nifi.processors.mqtt.common.StandardMqttMessage;
Expand All @@ -36,10 +40,13 @@

import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.TrustManagerFactory;
import javax.net.ssl.X509ExtendedKeyManager;
Expand All @@ -61,6 +68,13 @@ public HiveMqV5ClientAdapter(URI brokerUri, MqttClientProperties clientPropertie
this.logger = logger;
}

// Package-private constructor for injecting a test double for the underlying HiveMQ client.
HiveMqV5ClientAdapter(Mqtt5BlockingClient mqtt5BlockingClient, MqttClientProperties clientProperties, ComponentLog logger) {
this.mqtt5BlockingClient = mqtt5BlockingClient;
this.clientProperties = clientProperties;
this.logger = logger;
}

@Override
public boolean isConnected() {
return mqtt5BlockingClient.getState().isConnected();
Expand Down Expand Up @@ -127,30 +141,51 @@ public void publish(String topic, StandardMqttMessage message) {
}

@Override
public void subscribe(String topicFilter, int qos, ReceivedMqttMessageHandler handler) {
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.
public void subscribe(List<MqttTopicSubscription> subscriptions, ReceivedMqttMessageHandler handler) {
logger.debug("Subscribing to {}", subscriptions);

final List<Mqtt5Subscription> mqtt5Subscriptions = subscriptions.stream()
.map(subscription -> Mqtt5Subscription.builder()
.topicFilter(subscription.topicFilter())
.qos(Objects.requireNonNull(MqttQos.fromCode(subscription.qos())))
.build())
.collect(Collectors.toList());

final Mqtt5Subscribe mqtt5Subscribe = Mqtt5Subscribe.builder()
.addSubscriptions(mqtt5Subscriptions)
.build();

// Setting the "listener" callback is only possible with the async client, though sending the subscribe
// message should happen in a blocking way to make sure the processor is blocked until the ack arrives.
final CompletableFuture<Mqtt5SubAck> futureAck = mqtt5BlockingClient.toAsync().subscribe(mqtt5Subscribe, mqtt5Publish -> {
final ReceivedMqttMessage receivedMessage = new ReceivedMqttMessage(
mqtt5Publish.getPayloadAsBytes(),
mqtt5Publish.getQos().getCode(),
mqtt5Publish.isRetain(),
mqtt5Publish.getTopic().toString());
handler.handleReceivedMessage(receivedMessage);
});

final Mqtt5SubAck ack;
try {
final Mqtt5SubAck ack = futureAck.get(clientProperties.getConnectionTimeout(), TimeUnit.SECONDS);
logger.debug("Received mqtt5 subscribe ack: {}", ack);
ack = futureAck.get(clientProperties.getConnectionTimeout(), TimeUnit.SECONDS);
} catch (Exception e) {
throw new MqttException("An error has occurred during sending subscribe message to broker", e);
}
logger.debug("Received mqtt5 subscribe ack: {}", ack);

// A SUBACK carries one reason code per requested Topic Filter, in the order they were sent, so a subscription
// can be rejected individually, for example due to an ACL denial, while the others are granted.
final List<Mqtt5SubAckReasonCode> reasonCodes = ack.getReasonCodes();
final List<String> failedTopicFilters = new ArrayList<>();
for (int i = 0; i < reasonCodes.size() && i < subscriptions.size(); i++) {
if (reasonCodes.get(i).isError()) {
failedTopicFilters.add(subscriptions.get(i).topicFilter() + " (" + reasonCodes.get(i) + ")");
}
}
if (!failedTopicFilters.isEmpty()) {
throw new MqttException("Broker rejected subscription for the following topic filter(s): " + failedTopicFilters);
}
}

private static Mqtt5BlockingClient createClient(URI brokerUri, MqttClientProperties clientProperties, ComponentLog logger) throws TlsException {
Expand Down
37 changes: 33 additions & 4 deletions ...cessors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PahoMqttClientAdapter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,24 +20,32 @@
import org.apache.nifi.processors.mqtt.common.MqttClient;
import org.apache.nifi.processors.mqtt.common.MqttClientProperties;
import org.apache.nifi.processors.mqtt.common.MqttException;
import org.apache.nifi.processors.mqtt.common.MqttTopicSubscription;
import org.apache.nifi.processors.mqtt.common.ReceivedMqttMessage;
import org.apache.nifi.processors.mqtt.common.ReceivedMqttMessageHandler;
import org.apache.nifi.processors.mqtt.common.StandardMqttMessage;
import org.apache.nifi.ssl.SSLContextProvider;
import org.eclipse.paho.client.mqttv3.IMqttClient;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class PahoMqttClientAdapter implements MqttClient {

public static final int DISCONNECT_TIMEOUT = 5000;

// MQTT SUBACK reason code indicating the broker rejected the requested subscription, for example due to an ACL
// denial. Paho does not surface this as an error on its own, so the granted QoS array has to be checked here.
private static final int SUBACK_FAILURE_CODE = 0x80;

private final IMqttClient client;
private final MqttClientProperties clientProperties;
private final ComponentLog logger;
Expand All @@ -49,6 +57,14 @@ public PahoMqttClientAdapter(URI brokerUri, MqttClientProperties clientPropertie
client.setCallback(new DefaultMqttCallback());
}

// Package-private constructor for injecting a test double for the underlying Paho client.
PahoMqttClientAdapter(IMqttClient client, MqttClientProperties clientProperties, ComponentLog logger) {
this.client = client;
this.clientProperties = clientProperties;
this.logger = logger;
client.setCallback(new DefaultMqttCallback());
}

@Override
public boolean isConnected() {
return client.isConnected();
Expand Down Expand Up @@ -123,15 +139,28 @@ public void publish(String topic, StandardMqttMessage message) {
}

@Override
public void subscribe(String topicFilter, int qos, ReceivedMqttMessageHandler handler) {
logger.debug("Subscribing to {} with QoS: {}", topicFilter, qos);
public void subscribe(List<MqttTopicSubscription> subscriptions, ReceivedMqttMessageHandler handler) {
final String[] topicFilters = subscriptions.stream().map(MqttTopicSubscription::topicFilter).toArray(String[]::new);
final int[] qosLevels = subscriptions.stream().mapToInt(MqttTopicSubscription::qos).toArray();

logger.debug("Subscribing to {} with QoS: {}", Arrays.toString(topicFilters), Arrays.toString(qosLevels));

client.setCallback(new ConsumerMqttCallback(handler));

try {
client.subscribe(topicFilter, qos);
final IMqttToken token = client.subscribeWithResponse(topicFilters, qosLevels);
final int[] grantedQos = token.getGrantedQos();
final List<String> failedTopicFilters = new ArrayList<>();
for (int i = 0; i < grantedQos.length; i++) {
if (grantedQos[i] == SUBACK_FAILURE_CODE) {
failedTopicFilters.add(topicFilters[i]);
}
}
if (!failedTopicFilters.isEmpty()) {
throw new MqttException("Broker rejected subscription for the following topic filter(s): " + failedTopicFilters);
}
} catch (org.eclipse.paho.client.mqttv3.MqttException e) {
throw new MqttException("An error has occurred during subscribing to " + topicFilter + " with QoS: " + qos, e);
throw new MqttException("An error has occurred during subscribing to " + Arrays.toString(topicFilters) + " with QoS: " + Arrays.toString(qosLevels), e);
}
}

Expand Down
Loading
Loading