You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/05/07 14:12:49 UTC
[GitHub] [pulsar] eolivelli commented on a diff in pull request #15391: [enh] [broker] EntryFilter (PIP-105) - support per-Consumer filtering
eolivelli commented on code in PR #15391:
URL: https://github.com/apache/pulsar/pull/15391#discussion_r867356798
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java:
##########
@@ -147,13 +149,20 @@ public int filterEntriesForConsumer(Optional<EntryWrapper[]> entryWrapper, int e
msgMetadata = msgMetadata == null
? Commands.peekMessageMetadata(metadataAndPayload, subscription.toString(), -1)
: msgMetadata;
+ EntryFilter.FilterResult filterResult = EntryFilter.FilterResult.ACCEPT;
if (CollectionUtils.isNotEmpty(entryFilters)) {
- fillContext(filterContext, msgMetadata, subscription);
- if (EntryFilter.FilterResult.REJECT == getFilterResult(filterContext, entry, entryFilters)) {
+ fillContext(filterContext, msgMetadata, subscription, consumer);
+ filterResult = getFilterResult(filterContext, entry, entryFilters);
+ if (filterResult == EntryFilter.FilterResult.REJECT) {
entriesToFiltered.add(entry.getPosition());
entries.set(i, null);
entry.release();
continue;
+ } else if (filterResult == EntryFilter.FilterResult.RESCHEDULE) {
+ entriesToRedeliver.add((PositionImpl) entry.getPosition());
+ entries.set(i, null);
+ entry.release();
+ continue;
Review Comment:
This won't happen, because for each round we process the Entry only once, and it will be put in the set of messages to be redelivered. So we are not entering an infinite loop.
At the next round the Dispatcher will choose a Consumer (it may choose the same, if you configure priority for instance).
For the read entry requests to the bookies. Shouldn't we keep reading the entry from the local cache ?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java:
##########
@@ -147,13 +149,20 @@ public int filterEntriesForConsumer(Optional<EntryWrapper[]> entryWrapper, int e
msgMetadata = msgMetadata == null
? Commands.peekMessageMetadata(metadataAndPayload, subscription.toString(), -1)
: msgMetadata;
+ EntryFilter.FilterResult filterResult = EntryFilter.FilterResult.ACCEPT;
if (CollectionUtils.isNotEmpty(entryFilters)) {
- fillContext(filterContext, msgMetadata, subscription);
- if (EntryFilter.FilterResult.REJECT == getFilterResult(filterContext, entry, entryFilters)) {
+ fillContext(filterContext, msgMetadata, subscription, consumer);
+ filterResult = getFilterResult(filterContext, entry, entryFilters);
+ if (filterResult == EntryFilter.FilterResult.REJECT) {
entriesToFiltered.add(entry.getPosition());
entries.set(i, null);
entry.release();
continue;
+ } else if (filterResult == EntryFilter.FilterResult.RESCHEDULE) {
+ entriesToRedeliver.add((PositionImpl) entry.getPosition());
+ entries.set(i, null);
+ entry.release();
+ continue;
Review Comment:
This won't happen, because for each round we process the Entry only once, and it will be put in the set of messages to be redelivered. So we are not entering an infinite loop.
At the next round the Dispatcher will choose a Consumer (it may choose the same, if you configure priority for instance).
For the read entry requests to the bookies: shouldn't we keep reading the entry from the local cache ?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org