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]

Reply via email to