Repository navigation
PIP 105: Support pluggable entry filter in Dispatcher #12269
Description
Motivation
There are many users who need to use tag messages. The implementation of
this part has also been discussed before：
We need to provide a plug-in way so that developers can decide which messages can be sent to consumers and which ones are not.
In this way, not only tag messages can be supported, users can also use this extension point to implement other features.
Goal
In this PIP, we only implement the entry filter framework at the broker level, and we will continue to support the namespace and topic level in other PR.
API Changes And Implementation
Broker
add option entryFilterClassName in broker.conf and ServiceConfiguration, default value is null.
Add a factory class, if the entryFilterClassName is set, the factory class will be created, then init the filter
public interface EntriesFilterProvider {
// Use `EntriesFilterProvider` to create `EntriesFilter` through reflection
EntriesFilter createEntriesFilter(Subscription subscription);
// The default implementation class of EntriesFilterProvider
public DefaultEntriesFilterProviderImpl implements EntriesFilterProvider {
public DefaultEntriesFilterProviderImpl(ServiceConfiguration serviceConfiguration) {
//...
}
EntriesFilter createEntriesFilter(Subscription subscription) { //… }
}
}
Add some new interfaces to avoid relying on existing interfaces inside Pulsar. In this way, even if the internal interface changes in the future, the plugin can continue to run
// The interface that needs to be implemented in the plugin
public interface EntryFilter {
public FilterResult filterEntries(Entry entry, FilterContext context);
public static class FilterContext {
EntryBatchSizes batchSizes,
SendMessageInfo sendMessageInfo,
EntryBatchIndexesAcks indexesAcks,
ManagedCursor cursor,
boolean isReplayRead,
MessageMetadata msgMetadata
...
}
enum FilterResult {
ACCEPT, // deliver to the Consumer
REJECT, // skip the message
// ...
}
}
Why filter a single entry instead of List ?
The existing Dispatchers already traverse the entry list and parse the metadata of each Entry. We can reuse this data to avoid traversing the entry list twice and parsing the metadata twice.
Call the plugin in AbstractBaseDispatcher#filterEntriesForConsumer
if(entryFilter != null) {
FilterResult result = filterEntries(entry, FilterContext context);
if(FilterResult.REJECT.equals(result)) {
entries.set(i, null);
entry.release();
subscription.acknowledgeMessage(entry);
continue;
}
}
Client
Add optional attribute subscriptionProperties to the subscription, which can be set when subscribing to topic. These properties cannot be modified. We can only delete the subscription and create it again.
a. This attribute needs to be added to the subscribe command
b. For persistent subscriptions, these properties should stored in ZooKeeper with subscription
c. REST API (create-subscription) should also support setting subscriptionProperties
Example:
Consumer<byte[]> consumer = pulsarClient.newConsumer()
.topic(topic)
.subscriptionName("sub")
.subscriptionProperties(new HashMap<String, String>)
.subscribe();
Reject Alternatives
-
Plug-in Dispatcher
If we make this pluggable that we must define a limited private but
"stable" API. Enrico's suggestion is to define particular needs and then add features to make
pluggable single specific parts of the dispatcher. -
Create a message dimension filter
At this stage, we can only do Entry-level filtering. If the Message in the Entry is forced to be filtered on the Broker side, there will be problems in the subsequent consumer ack.
Therefore, if we want to use this filter, we must set enableBatching=false, which is the same as delayed messages.
Activity
eolivelli commentedon Oct 4, 2021
It is not clear to me if the Consumer can attach some configuration value (like JMS message selector) that can be used in the Filter.
Otherwise we can only create static filters
315157973 commentedon Oct 5, 2021
We can use it like this:
Set properties when creating consumer.
entryFilterClasscan be defined as an adapter, which internally determines what filter logic to use according to properties315157973 commentedon Oct 8, 2021
@eolivelli Do you think this is okay? If there is no problem, I will initiate a vote
eolivelli commentedon Oct 8, 2021
@315157973
I am okay as far there is an API for the client to set such properties and to be able to use them in the broker.
can you please give me an example about how to set custom properties in the subscription in a way that the broker can leverage those configurations ?
315157973 commentedon Oct 11, 2021
@eolivelli I added a description of this part. PTAL, Thanks
315157973 commentedon Oct 12, 2021
@eolivelli PTAL, thanks
eolivelli commentedon Oct 12, 2021
I would change 'subscriptProperties' to 'subscriptionProperties'
Apart from this LGTM
eolivelli commentedon Oct 12, 2021
What happens if I subscribe from two different consumer with different subscriptProperties ?
315157973 commentedon Oct 12, 2021
subscriptionPropertiesis immutable. Attempts to modify will return an exception. If Properties is empty or the same as the existing one, we can successfully subscribe.wangjialing218 commentedon Oct 15, 2021
Could user implement their feature and deside which messages can be sent to consumers according to deserialized entry data?
For interface
public FilterResult filterEntries(Entry entry, FilterContext context);Entryparameter have no meaning for user unless user deserializeEntry.getData()in there implement, which may bring more CPU and GC workload on the broker.So should we allow user to deserialize each entry data in this interface?
yangl commentedon Oct 26, 2021
@315157973 Did this support the topic with end-to-end encryption enabled?
315157973 commentedon Oct 26, 2021
FilterContext, and you don't even need to re-parse the metadata. The incoming parameter is an Entry, you can play freely. You can even modify the data in Entry.315157973 commentedon Oct 26, 2021
It depends on your implementation class. End-to-End encryption will also go through this logic
wangjialing218 commentedon Oct 26, 2021
If user want to play anything with
Entryparameter, user need to deserialize entire Entry to heap in their implement, then user could modify the data in Entry.As @codelipenghui suggested in #11962 (comment), we should avoid to deserialize Entry to heap which will bring more GC workload on the broker.
If we do not want user to do this, we could remove the Entry parameter.
3 remaining items
wangjialing218 commentedon Oct 26, 2021
The Entry data here is in direct memory, if user call
Entry.getData()in there implmentation, the Entry data will transfer to heap. If message traffic is high, broker heap may be full soon and GC will repleated in seconds.We need to evaluate the impact carefully.
kuangye098 commentedon Nov 3, 2021
I recommend:
[PIP-105] Part-1 Support add subscription properties (#12869)
Anonymitaet commentedon Nov 26, 2021
Hi @315157973 for the doc side, I've discussed with @codelipenghui, he suggests creating an independent chapter
message filterunderDevelopmentand considering addingpluginandprotocol handlersections under that chapter.[PIP-105] Part-1 Support add subscription properties (apache#12869)
[PIP-105] Part-1 Support add subscription properties (apache#12869)
[-]Support pluggable entry filter in Dispatcher[/-][+]PIP 105: Support pluggable entry filter in Dispatcher[/+]lhotari commentedon Jan 20, 2022
For adding some more context, there was a proposal in the past from @andrekramer1 , that was discussed in this dev mailing list thread: https://lists.apache.org/thread/k0hb33nz42f3603hvs03p7lmltjgj40z .
I would prefer a standard feature approach in Pulsar instead of making it an optional plugin. The reason for this is to prevent fragmentation of Pulsar into a set of optional plugins which might or might not be available or installed.
github-actions commentedon Feb 28, 2022
The issue had no activity for 30 days, mark with Stale label.
asafm commentedon Jul 13, 2022
I can't find the vote in the mailing list searching PIP-105.
eolivelli commentedon Jul 13, 2022
@asafm good call! I can't find it either
ullebe1 commentedon Jul 22, 2024
Can you elaborate on the reason for this please? For long-lived subscriptions on persistent topics being able to modify the properties could definitely be beneficial, as recreating the subscription could be quite cumbersome if done regularly.
merlimat commentedon Oct 24, 2026
Thanks @315157973 for PIP-105. The vote passed (https://lists.apache.org/thread/bdw6cnn9hdmljxf4751jl76c4g47ydny), and the entry filter framework is on master through #12869 and #12970 (
EntryFilter, configured withentryFilterNamesandentryFiltersDirectory); the proposal is at https://github.com/apache/pulsar/blob/master/pip/pip-105.md. Closing as completed.