[feat][pip] PIP-419: Support consumer message filter on broker-side for specific subscription

[AuroraTwinkle]

Fixes #xyz

Main Issue: #8544
#7629

PIP: #xyz

Motivation

Currently, Pulsar does not have the ability to filter messages by subscription, this pip try to implement message filtering on the pulsar broker side, enrich the pulsar ecosystem, save bandwidth, and help RocketMQ users switch to Pulsar more smoothly. And, I've noticed two similar PRs #8544 and #7629, but neither was approved, so I want to try to get this work implemented.

Modifications

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository:
AuroraTwinkle#6

[github-actions]

@AuroraTwinkle Please add the following content to your PR description and select a checkbox:

- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

[AuroraTwinkle]

@lhotari @StevenLuMT @liangyepianzhou I have submitted a pip about message filtering. Would you please review it when you have time? Looking forward to your suggestions. Thanks!

[liangyepianzhou]

Excellent work! I think this feature is indeed useful. However, I suggest that the filter expressions should be configured, modified, and deleted through the management/control plane (Pulsar Admin/CLI), rather than being set during consumer initialization. My reasoning is as follows:

1. Practical Operational Considerations: When maintaining a messaging service, we typically provide users with a management platform to configure subscription settings. If filter conditions are set during consumer initialization, any configuration changes would require propagating updates to user clients and forcing client reinitialization to take effect. This prevents direct updates via admin operations.

2. Service Stability: Broker-side configuration changes requiring client reinitialization could introduce consumption service instability.

3. Subscription Consistency: A single subscription may have multiple consumer instances. Setting filter conditions at consumer initialization risks inconsistent filtering rules across consumers under the same subscription, potentially causing rule overwrites (e.g., Consumer A sets Filter X while Consumer B sets Filter Y, leading to conflicts).

Regarding point 3: This mirrors real-world issues observed in RocketMQ's message filtering. I've encountered multiple cases where tag-based filtering failed because different consumers in the same consumer group subscribed to conflicting tags, resulting in rule overrides.

Recommendation: Store filter rules as subscription metadata (managed via Admin API/CLI) rather than per-consumer settings. This would:

• Ensure consistent filtering across all consumers in a subscription
• Allow dynamic rule updates without client restarts
• Prevent rule conflicts between consumers

[AuroraTwinkle]

It is a good idea to configure the filtering rules through admin API. This can avoid the problems you mentioned, which is also the current problem of RocketMQ. I will refactor it together after receiving more comments.

[lhotari]

It's already possible to handle filtering rules through the Admin API. The pulsar-admin CLI tool has a plugin model that allows adding also new commands. This approach is used in Pulsar JMS: https://github.com/datastax/pulsar-jms/blob/master/pulsar-jms-admin-ext/src/main/java/com/datastax/oss/pulsar/jms/cli/SubscriptionBaseCommand.java (more context in my other comment, #24305 (comment))

[lhotari]

In Pulsar JMS, you could also provide the filter at a consumer level with consumer properties. If we were to add message filtering as a top level feature, it would be great to have full support in the clients too. In my other comment, I commented about this aspect

One of the possible compromises could be to make it an optional feature. In the Pulsar Java client API, we'd need to have a way to handle optional features in the builder interface itself so that the main builder interface for consumers doesn't get distracted by options that are only supported in certain configurations. One way to do this is with an "extensions" pattern. Where the builder would have a method extension(Class consumerExtensionBuilderClass) which continues configuring the extension in a "nested" builder.

[lhotari]

And, I've noticed two similar PRs #8544 and #7629, but neither was approved, so I want to try to get this work implemented.

@AuroraTwinkle For former context why such feature has been rejected in the past, this email thread is a good read: https://lists.apache.org/thread/k0hb33nz42f3603hvs03p7lmltjgj40z .

There simply hasn't been consensus that we'd like to move Pulsar together as a community to this direction.
As a compromise, it was made possible to do broker side filtering using custom plugins with "PIP 105: Support pluggable entry filter in Dispatcher". and related subscription properties #15503. There's an implementation in Pulsar JMS using this approach: https://github.com/datastax/pulsar-jms/blob/master/pulsar-jms-filters/src/main/java/com/datastax/oss/pulsar/jms/selectors/JMSFilter.java

There are some performance challenges currently when using broker side filtering. However, many of the problems have been addressed over time and if we'd introduce the feature as a top level feature, there would be more motivation to address more of the challenges such as ones of the gaps in broker cache and handling of "ack holes" in Pulsar efficiently.

The "ack hole" challenge is similar for delayed message delivery at large scale. Ack hole handling on broker side has been optimized with changes such as #9292, (must be enabled with managedLedgerPersistIndividualAckAsLongArray=true since default is false after #23759), increasing managedLedgerMaxUnackedRangesToPersist, setting managedCursorInfoCompressionType=LZ4, etc.

For broker cache, there's lot more to do. #23466 .
One of the workarounds is cacheEvictionByMarkDeletedPosition=true for message filtering. I've have a WIP branch to address many of the problems, but haven't had a chance to push that forward.
It's just an example that once there's a requirement to address performance issues, those would get addressed over time. That's why I'd like to see message filtering being added as a top level feature in Pulsar in the long term.

One of the possible compromises could be to make it an optional feature. In the Pulsar Java client API, we'd need to have a way to handle optional features in the builder interface itself so that the main builder interface for consumers doesn't get distracted by options that are only supported in certain configurations. One way to do this is with an "extensions" pattern. Where the builder would have a method <T extends ConsumerExtensionBuilder> extension(Class<T> consumerExtensionBuilderClass) which continues configuring the extension in a "nested" builder.

[lhotari]

Pulsar JMS reuses ActiveMQ's SQL-92 parser for filtering messages
https://github.com/datastax/pulsar-jms/blob/master/pulsar-jms-filters-common/src/main/java/com/datastax/oss/pulsar/jms/selectors/SelectorSupport.java

[lhotari]

The project has forked the filtering logic from ActiveMQ in this module: https://github.com/datastax/pulsar-jms/tree/master/activemq-filters

[lhotari]

This would be terrible from security perspective. Something like Google CEL or Spring Expression Language (SpEL) would be a better choice.

[dao-jun]

I’m +1 with the motivation.

even though we have EntryFilter already, but it's not ready to use right out of the box.

In the details, maybe we can impl EntryFilter, and manage the rules via pulsar admin?

SpEL is a better choice than groovy scripts, it can easily lead to metadata area OOM because of caching classloader, I had encountered this problem in our business systems.

Besides, I hope you can consider that the ack hole problems, and it could comes incorrect stats/metrics(such as backlog).

[AuroraTwinkle]

And, I've noticed two similar PRs #8544 and #7629, but neither was approved, so I want to try to get this work implemented.

@AuroraTwinkle For former context why such feature has been rejected in the past, this email thread is a good read: https://lists.apache.org/thread/k0hb33nz42f3603hvs03p7lmltjgj40z .

There simply hasn't been consensus that we'd like to move Pulsar together as a community to this direction. As a compromise, it was made possible to do broker side filtering using custom plugins with "PIP 105: Support pluggable entry filter in Dispatcher". and related subscription properties #15503. There's an implementation in Pulsar JMS using this approach: https://github.com/datastax/pulsar-jms/blob/master/pulsar-jms-filters/src/main/java/com/datastax/oss/pulsar/jms/selectors/JMSFilter.java

There are some performance challenges currently when using broker side filtering. However, many of the problems have been addressed over time and if we'd introduce the feature as a top level feature, there would be more motivation to address more of the challenges such as ones of the gaps in broker cache and handling of "ack holes" in Pulsar efficiently.

The "ack hole" challenge is similar for delayed message delivery at large scale. Ack hole handling on broker side has been optimized with changes such as #9292, (must be enabled with managedLedgerPersistIndividualAckAsLongArray=true since default is false after #23759), increasing managedLedgerMaxUnackedRangesToPersist, setting managedCursorInfoCompressionType=LZ4, etc.

For broker cache, there's lot more to do. #23466 . One of the workarounds is cacheEvictionByMarkDeletedPosition=true for message filtering. I've have a WIP branch to address many of the problems, but haven't had a chance to push that forward. It's just an example that once there's a requirement to address performance issues, those would get addressed over time. That's why I'd like to see message filtering being added as a top level feature in Pulsar in the long term.

One of the possible compromises could be to make it an optional feature. In the Pulsar Java client API, we'd need to have a way to handle optional features in the builder interface itself so that the main builder interface for consumers doesn't get distracted by options that are only supported in certain configurations. One way to do this is with an "extensions" pattern. Where the builder would have a method <T extends ConsumerExtensionBuilder> extension(Class<T> consumerExtensionBuilderClass) which continues configuring the extension in a "nested" builder.

Thank you Lari for your rigorous and detailed suggestions. Please give me some time to understand them carefully.

[AuroraTwinkle]

I’m +1 with the motivation.

even though we have EntryFilter already, but it's not ready to use right out of the box.

In the details, maybe we can impl EntryFilter, and manage the rules via pulsar admin?

SpEL is a better choice than groovy scripts, it can easily lead to metadata area OOM because of caching classloader, I had encountered this problem in our business systems.

Besides, I hope you can consider that the ack hole problems, and it could comes incorrect stats/metrics(such as backlog).

Thanks.I will consider SpEL and ack hole problems.