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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
