codelipenghui commented on code in PR #15391:
URL: https://github.com/apache/pulsar/pull/15391#discussion_r867858865
##########
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.
Yes, it's not an infinite loop for one dispatch round, but the message will
be redelivered again and again until there is a consumer who can accept the
message right? Just to make sure I understand it correctly.
> For the read entry requests to the bookies: shouldn't we keep reading the
entry from the local cache ?
We don't have a catchup read cache for now. It's a different story, the
client-side redelivery might also read the entry from the bookies, so it's not
a blocker.
--
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]
