Add FilteringAdapter in SQS

[imsosleepy]

📢 Type of change

• Bugfix
• New feature
• Enhancement
• Refactoring

📜 Description

Added support for filtering SQS messages before they reach the listener using a MessageFilter.

The filter can be configured through SqsMessageListenerContainerFactory, and only messages that match the condition will be passed to the listener.

💡 Motivation and Context

Previously, all incoming messages were forwarded to the listener, requiring manual filtering within business logic.

This change enables declarative filtering at the container level, helping reduce unnecessary processing and keeping listener logic clean.

💚 How did you test it?

Added integration test SqsMessageFilterIntegrationTests to verify that filtered messages are not delivered to the listener.

Confirmed that only messages satisfying the filter are passed through and acknowledged.

📝 Checklist

• I reviewed submitted code
• I added tests to verify changes
• I updated reference documentation to reflect the change
• All tests passing
• No breaking changes

🔮 Next steps

This initial implementation supports filtering on a single message basis only, and does not yet account for batch or asynchronous filtering scenarios.

As suggested in #1373, we plan to extend support to:

• Batch message filtering
• Asynchronous filtering
• Support for filtering based on fields other than the message payload
  (e.g., message attributes, custom headers, or system metadata)

[MatejNedic]

Hey @imsosleepy tnx on making the PR! Will check PR next week since I am on the conference this week and Tomaz is on vacation.

[MatejNedic]

Hey @imsosleepy,
Tnx on PR to me this looks good.

I left 2 comments neat picks nothing big and have few questions.

I would like to see more in terms of filter, this can be done later or in this PR. Firstly what I would like to ask if you are okey on working on this in this PR to extend it or we tackle this in new PR? :)

Second @tomazfernandes please check if you agree with it.

1. I think we should support something like this.
   @SqsListener(filterMethod = "filterMethodBean")

2. @tomazfernandes just to make sure since this is happening inside container and we are actually not allowing messageListener impl call and ack will be autoamatic in default case?

[MatejNedic]

Neat pick again but please switch to for. Since size will be always maximum 10 lets switch to for loop to take benefit of performance.

[tomazfernandes]

Thanks @imsosleepy for the changes and @MatejNedic for looking into it!

The MessageFilter interface looks good.

I'm not sureabout filtering messages at the message listener level since we'd still have the message go through the interceptors and error handlers.

I believe it would make more sense to filter it at the Sink level, so the message never gets emitted and doesn't go through the other components, we just acknowledge it.

What do you folks think?

[MatejNedic]

@tomazfernandes I agree makes more sense to me as well. Abstraction MessageFilter is spot on what I would expect.

What about @SqsListener(messageFilter = "messageFilterBeanMethod" ) ? I think this makes much sense I see few use cases.

Are you sure message will go through error handler if it ends without exception in filter tho?

[tomazfernandes]

What about @SqsListener(messageFilter = "messageFilterBeanMethod" ) ? I think this makes much sense I see few use cases.

There's nothing really you can do through the annotation that you can't do with the ConfigurationOptions. In general, my rule of thumb is to limit the amount of configurations on the annotation unless users explicitly ask for them, as otherwise we might end up with all configurations that are available on ContainerOptions in the annotation and that would get messy.

For this particular case, I would suggest we stick with the ContainerOptions for the scope of this PR to keep it simpler, and we can open an issue / separate PR to add the annotation configuration after we merge it if we think that's important.

What do you think, makes sense?

[tomazfernandes]

Are you sure message will go through error handler if it ends without exception in filter tho?

Yeah, if it doesn't return an error it won't go through the error handler, but if it does return an error, we'll now have a message we don't want to process in the error handler, or perhaps no message at all which would likely trigger another error.

If we prevent the message from entering the processing pipeline altogether I believe it'll be simpler to manage the outcomes.

[imsosleepy]

Thanks so much for the detailed feedback @tomazfernandes and @MatejNedic!

I agree that filtering at the Sink level makes a lot more sense and helps avoid unnecessary processing through interceptors and error handlers.

That said, I’m still wrapping my head around the architecture around MessageSink and how exactly to plug in filtering at that level. If you have a more concrete suggestion or example of where the filtering logic should live and how it should be wired into the container, that would help me move faster and avoid misalignment.

I’ll dig a bit deeper on my end in the meantime. Thanks again for the guidance — really appreciate it!

[imsosleepy]

As suggested, I’ve moved the message filtering logic to the MessageSink layer to avoid unnecessary processing through interceptors and error handlers.

Before proceeding with additional test for batch and FIFO modes, I’d appreciate a review of the current structure to ensure we're aligned.

[MatejNedic]

LGTM flexible easy to use I like it!

[MatejNedic]

Hey @tomazfernandes can you check this PR when you get time?

[tomazfernandes]

Hey @imsosleepy @MatejNedic, I apologize for the delay.

I left a few comments, let me know your thoughts!

[tomazfernandes]

On a second thought, I think the interface should receive and return a Collection of messages rather than a single one.

Something like:

public interface MessageFilter<T> {
    Collection<Message<T>> filter(Collection<Message<T>> messages);
}

The reason is that if the filter includes some I/O (e.g. querying the DB to check if the users exist, or making an http request), that'll be more performant as a batch query than in a for loop, and if the user is receiving e.g. 500 messages in a batch that can make a sizeable difference.

If the user wants to filter individual messages, it's simple enough for them to iterate on the collection themselves.

What do you folks think?

[tomazfernandes]

The MessageSink is a composable component, so we should favor composition over inheritance.

In practical terms, it can extend from AbstractDelegatingMessageListeningSinkAdapter instead - see MessageGroupingSinkAdapter for an example and how it's assembled in the ContainerFactory.

Makes sense?

[tomazfernandes]

This part of the code needs to be non-blocking, otherwise it may block the polling thread e.g. if the filter makes a request or DB query.

I'm not 100% sure what's the best approach, but we could:

• Add the filtering logic in the execute method in AbstractMessageProcessingPipelineSink in a way that filtering happens in a separate thread
• Make the filtering interface non-blocking (CompletableFuture-based)

I would probably go with the first approach.

For that, we'd probably need to have the MessageFilter in the AbstractMessageProcessingPipelineSink directly.

Let me know your thoughts.

[imsosleepy]

Hey @imsosleepy @MatejNedic, I apologize for the delay.

I left a few comments, let me know your thoughts!

@tomazfernandes Thanks for the suggestions — I agree with the points you’ve raised on composition over inheritance and non-blocking filtering. I’ll work on these changes over the weekend and share the results. The non-blocking filtering part in the BatchMessageSink area might need a bit more input from you as I work through the details.

[imsosleepy]

@tomazfernandes I’d like to get feedback on whether my current approach makes sense, although FIFO is not yet working.

1. I felt it was too heavy to pass options all the way down to the execute level, so I implemented an asyncFilter.
2. While working on this, I found that wrapping OrderedMessageSink’s execute with asyncFilter caused issues — the ordering broke and some tests started failing.
3. It seems possible to set the filter in AbstractPipelineMessageListenerContainer’s configureMessageSink. This way, the filter can be configured at a higher level without pushing it all the way down to execute. However, if we do pass it down to execute, improving execute might be a better approach, because:
4. If we keep the current structure, the doEmit implementation in OrderedMessageSink will become quite complex.

In summary, the main question is whether the filter should be passed down to execute.

I’d appreciate any advice from others.