NIFI-16113 Adds new processor for changing visibility timeouts of mes…

[pkelly-nifi]

…sages from SQS

Summary

NIFI-16113

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

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

Documentation

• Documentation formatting appears as expected in rendered files

[exceptionfactory]

Thanks for proposing this new Processor @pkelly-nifi.

On initial review, it seems like a narrow use case that is not well suited for inclusion as part of the standard project distribution.

To understand the use case a bit more, is setting the visibility timeout that should be included on the PutSQS Processor as an optional property, as opposed to a separate Processor?

[pkelly-nifi]

Hi @exceptionfactory, thank you for your input. Admittedly this is a narrow use case, but it can be important.

To answer your question, setting the visibility timeout in PutSQS processor can't be done. There is no option in the SQS API for setting the visibility timeout on the message during the SendMessage operation, so it's not something we could incorporate into PutSQS even if we wanted to. You can set a message delay when sending a message, but not a visibility timeout. The difference is that a delay affects the time between when you send the message and when a client can first receive that message, whereas the visibility timeout is the interval between when a client first receives the message (but does not delete it) and when the message is considered to be a failure and SQS automatically replays it to another client. It will continue to be replayed at the visibility timeout interval until it is deleted by a client or considered to be a dead letter by the service, based on the configuration of the queue.

Visibility timeouts can be set in three ways:

1. It can be set when you receive the message. NiFi currently does this in GetSQS during the ReceiveMessage call, before you know anything about the message, so all messages retrieved by a single GetSQS use a common visibility timeout.
2. The timeout can be updated after receiving the message in a ChangeMessageVisibilityTimeout call. NiFi does not currently support this, but this new processor does.
3. A default visibility timeout can be set for the queue, but that doesn't really affect NiFi since the default visibility timeout is overwritten by the GetSQS processor when the message is retrieved.

I think there are two important use case for this processor.

First, being able to set a per-message visibility timeout for S3 event notifications, which something NiFi cannot currently do. Let's say you have an S3 bucket that normally receives a mix of small files (on the order of a few MBs), and very large files (hundreds of GBs or into TBs), and you have S3 set up to send you event notifications to SQS for objects that are uploaded. You drive the flow for processing those files using a GetSQS processor to receive these event notifications. You want to be able to very quickly retry any small files that fail, but you need to allow a lot of time for processing the big files. Amazon's documentation recommends calling ChangeMessageVisibilityTimeout in order to set appropriate visibility timeouts for each message, depending on processing needs:
https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html

This new processor would allow a flow designer to follow Amazon's recommendations, and allow for better retry performance for the mixed data sizes. Currently, the flow designer would need to configure GetSQS to set a visibility timeout large enough to cover the slowest file, which for this case would be several hours. Small files that fail would need to wait this long interval until their notifications are replayed.

What this new processor would allow is to set the visibility timeout in GetSQS to only a few minutes to cover the expected small files. You would then feed the event notification messages into EvaluateJsonPath to extract the object's size from the message, then evaluate the size and optionally use this new processor to set a new visibility timeout for messages for large objects (SQS supports up to 12 hours) in order to give you enough time to process them. Messages for small objects would replay on the shorter interval while allowing you enough time to process the large files without their messages getting replayed.

There are several other cases that are similar, but they are all derived from this same concept: you want to be able to set different replay intervals based on some characteristic of the message, which is something NiFi cannot currently do.

The second important case is about efficiently recovering from errors. This new processor could be called to set a visibility timeout of "0 seconds", which would cause SQS to immediately replay the message to trigger the initial flow again. This is useful if, for example, you know the file has failed in a way that requires processing it again from the beginning, but you don't want to wait for the original visibility timeout that GetSQS set to expire. This is especially useful if the messages come from a queue to which you do not have SendMessage permission since you can't just resend the message to the queue in those cases, but updating the timeout to zero would allow the original message to be replayed.

I'm happy to discuss further, but I think this could be a handy, if not critical, feature for some flows. I agree that the cases are pretty narrow, but they're important and currently cannot be done inside of NiFi without significant complication.