NIFI-16203 Support multiple Topic Filters in ConsumeMQTT

[drewyh999]

Summary

NIFI-16203

ConsumeMQTT exposes a single Topic Filter property and issues one subscription per connection, so consuming from N topics requires N Processor instances, each opening its own broker connection with its own Client ID. A wildcard filter is the usual workaround, but it is unusable when topic names are flat or externally dictated, or when broker ACLs authorise only explicit topics.

The MQTT SUBSCRIBE packet natively carries a list of (Topic Filter, QoS) pairs on every protocol version NiFi supports, so Topic Filter now accepts a comma separated list and every filter is subscribed to with a single SUBSCRIBE request over one connection.

Backward compatibility

The property is not renamed and a value containing no comma is treated as a single Topic Filter and used verbatim, so existing flows are unaffected and no property migration is needed. Only the entries of a comma separated value are trimmed, since leading and trailing whitespace is significant in an MQTT Topic Filter.

Because MQTT Topic Filters may legally contain a comma, a filter that itself contains one is interpreted as multiple filters. This is called out in both the property description and additionalDetails.md.

SUBACK handling

While implementing this it became apparent that the SUBACK response was only logged, so a Topic Filter rejected by the broker (for example due to an ACL denial) failed silently while the Processor still looked healthy. This matters more with multiple filters, since a SUBSCRIBE listing several filters can be granted partially. Both adapters now inspect the per filter response — HiveMQ through the SUBACK reason codes, Paho through the granted QoS array — and fail with the offending Topic Filter listed.

Since a partial grant leaves the client connected and subscribed, a failure now disconnects and closes it rather than only dropping the reference, which previously leaked the broker connection and left an orphaned client feeding the internal queue.

Changes

• Add MqttTopicSubscription and change MqttClient.subscribe to take a list
• Apply the shared subscription prefix ($share/<Group ID>/) to each filter individually
• Reject blank and duplicate Topic Filters during validation
• Report every subscribed Topic Filter in the aggregated provenance transit URI
• Stop the HiveMQ adapter from re-wrapping its own rejection exception, which hid the rejected Topic Filter behind a generic message

Tracking

Please complete the following tracking steps prior to pull request creation.

Issue Tracking

• Apache NiFi Jira issue created

Pull Request Tracking

• Pull Request title starts with Apache NiFi Jira issue number, such as NIFI-00000
• Pull Request commit message starts with Apache NiFi Jira issue number, as such NIFI-00000
• Pull request contains commits signed with a registered key indicating Verified status

Pull Request Formatting

• Pull Request based on current revision of the main branch
• Pull Request refers to a feature branch with one commit containing changes

Verification

Please indicate the verification steps performed prior to pull request creation.

Build

• Build completed using ./mvnw clean install -P contrib-check
  • JDK 21
  • JDK 25

Verified on the nifi-mqtt-processors module only (JDK 21): ./mvnw -pl nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors test checkstyle:check — 57 tests pass, checkstyle clean. A full contrib-check build across both JDKs has not been run locally; happy to follow up on any CI findings.

Licensing

• New dependencies are compatible with the Apache License 2.0 according to the License Policy
• New dependencies are documented in applicable LICENSE and NOTICE files

No new dependencies are introduced.

Documentation

• Documentation formatting appears as expected in rendered files

additionalDetails.md for ConsumeMQTT gains a "Multiple Topic Filters" section, and the Topic Filter property description is updated.