drewyh999 opened a new pull request, #11556: URL: https://github.com/apache/nifi/pull/11556
# Summary [NIFI-16203](https://issues.apache.org/jira/browse/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](https://issues.apache.org/jira/browse/NIFI-16203) 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](https://docs.github.com/en/authentication/managing-commit-signature-verification/signing-commits) 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 - [x] New dependencies are compatible with the [Apache License 2.0](https://apache.org/licenses/LICENSE-2.0) according to the [License Policy](https://www.apache.org/legal/resolved.html) - [x] New dependencies are documented in applicable `LICENSE` and `NOTICE` files No new dependencies are introduced. ### Documentation - [x] 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
