[Bug] An exception while filtering a read stops dispatch on a Shared subscription until the topic unloads #27218

Search before reporting

• I searched in the issues and found nothing similar. The closest is Any runtime exception in AbstractBaseDispatcher breaks the broker #12405 (2021), closed on 2026-10-24 because it was reported on a release before 3.0, with an invitation to open a new issue if it still happens on a supported release. It does: reproduced below on 5.0.0, and the code path is unchanged on master (7d0c964).

Read release policy

• I understand that unsupported versions don't get bug fixes.

User environment

apachepulsar/pulsar:5.0.0 standalone with default settings plus one broker-level entry filter (entryFiltersDirectory, entryFilterNames). Shared subscription, bin/pulsar-client 5.0.0 producing with batching disabled and consuming.

Issue Description

An exception thrown while PersistentDispatcherMultipleConsumers filters or sends a read is not handled. After one such exception the subscription delivered nothing more, neither that read's entries nor a message published afterwards, until the topic was unloaded.

With an entry filter that rejects m3 and throws once on m5 (reproduction below):

1. m1..m8 are published before the consumer connects, so they arrive in one read. The filter throws on m5; OpReadEntry logs readEntriesComplete failed with nextReadPosition: 5:8, past all eight.
2. 20 s after a Shared consumer connects it has received nothing; backlog 8. m3, rejected before the throw, was not acked either.
3. m9, published with that consumer still connected, is not delivered; backlog 9.
4. pulsar_broker_pending_bytes_to_dispatch stays at 352.0, the read's size.
5. After pulsar-admin topics unload, the consumer receives m1 m2 m4 m5 m6 m7 m8 m9 (the filter now accepts m5), m3 is rejected and acked, backlog 0.

The same steps give the same result with dispatcherDispatchMessagesInSubscriptionThread=true; the exception then ends on the dispatch executor (SingleThreadExecutor - Error while running task) instead of in OpReadEntry.

Why, with line numbers from the v5.0.0 tag:

• readEntriesComplete (PersistentDispatcherMultipleConsumers.java:692) clears havePendingRead first, then handleSendingMessagesAndReadingMore (:745) calls sendMessagesToConsumers (:787), whose try has only a finally clearing sendInProgress. A throw from filterEntriesForConsumer (:882) or a consumer's sendMessages (:887, :994) skips the readMoreEntries() that would follow, and OpReadEntry.completeNow (OpReadEntry.java:377) only logs it; readEntriesFailed is not called. No read is outstanding and nothing schedules one, so new entries do not wake the dispatcher either (step 3).
• The read's entries are past the cursor's read position, in no consumer's pending acks, and for a replay read already removed from redeliveryMessages (:873). Only a rewind or an unload brings them back.
• filterEntriesForConsumer (AbstractBaseDispatcher.java:126) nulls and releases a REJECTed or RESCHEDULEd entry inside its loop (:166, :177) but acks or redelivers it only after the loop (:301, :314), so a throw on a later entry loses that settlement (m3 in step 2).
• updatePendingBytesToDispatch(-totalBytesSize) (:750) runs after sendMessagesToConsumers returns, so the gauge keeps the read's bytes (step 4).

An entry filter plugin is one way to throw there; metadata parsing of a corrupt entry is another (the stack in #12405).

Error messages

ERROR org.apache.bookkeeper.mledger.impl.OpReadEntry - readEntriesComplete failed {lastPosition=5:7, op=public/default/persistent/repro sub{ readPosition: 5:0, nextReadPosition: 5:8, maxPosition: 9223372036854775807:9223372036854775807, entries count: 8, count: 500 }}
java.lang.IllegalStateException: repro: the entry filter threw once
	at repro.ThrowOnceEntryFilter.filterEntry(ThrowOnceEntryFilter.java:27)
	at org.apache.pulsar.broker.service.plugin.EntryFilterWithClassLoader.filterEntry(EntryFilterWithClassLoader.java:46)
	at org.apache.pulsar.broker.service.EntryFilterSupport.getFilterResult(EntryFilterSupport.java:85)
	at org.apache.pulsar.broker.service.EntryFilterSupport.runFiltersForEntry(EntryFilterSupport.java:66)
	at org.apache.pulsar.broker.service.AbstractBaseDispatcher.filterEntriesForConsumer(AbstractBaseDispatcher.java:164)
	at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.trySendMessagesToConsumers(PersistentDispatcherMultipleConsumers.java:882)
	at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.sendMessagesToConsumers(PersistentDispatcherMultipleConsumers.java:793)
	at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.handleSendingMessagesAndReadingMore(PersistentDispatcherMultipleConsumers.java:748)
	at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.readEntriesComplete(PersistentDispatcherMultipleConsumers.java:741)
	at org.apache.bookkeeper.mledger.impl.OpReadEntry.completeNow(OpReadEntry.java:377)
	at org.apache.bookkeeper.mledger.impl.OpReadEntry.completeWithDepthLimit(OpReadEntry.java:342)
	at org.apache.bookkeeper.mledger.impl.OpReadEntry.complete(OpReadEntry.java:326)

Reproducing the issue

An entry filter that rejects filter=reject and throws once, on the first filter=throw-once it sees:

package repro;

import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.pulsar.broker.service.plugin.EntryFilter;
import org.apache.pulsar.broker.service.plugin.FilterContext;
import org.apache.pulsar.common.api.proto.KeyValue;

public class ThrowOnceEntryFilter implements EntryFilter {

    private static final AtomicBoolean THROWN = new AtomicBoolean();

    @Override
    public FilterResult filterEntry(Entry entry, FilterContext context) {
        for (KeyValue property : context.getMsgMetadata().getPropertiesList()) {
            if (!"filter".equals(property.getKey())) {
                continue;
            }
            if ("reject".equals(property.getValue())) {
                return FilterResult.REJECT;
            }
            if ("throw-once".equals(property.getValue()) && THROWN.compareAndSet(false, true)) {
                throw new IllegalStateException("repro: the entry filter threw once");
            }
        }
        return FilterResult.ACCEPT;
    }

    @Override
    public void close() {
    }
}

Compiled against the jars in /pulsar/lib of the image and packed as throw-once.nar with META-INF/bundled-dependencies/throw-once-filter.jar and META-INF/services/entry_filter.yml:

name: throw-once
description: Rejects filter=reject, throws once on the first filter=throw-once
entryFilterClass: repro.ThrowOnceEntryFilter

Then, with the NAR in ./filters:

docker run -d --name repro -v "$PWD/filters":/filters:ro \
  -e PULSAR_PREFIX_entryFiltersDirectory=/filters \
  -e PULSAR_PREFIX_entryFilterNames=throw-once \
  apachepulsar/pulsar:5.0.0 \
  bash -c 'bin/apply-config-from-env.py conf/standalone.conf && exec bin/pulsar standalone -nfw -nss'
# wait for: docker exec repro bin/pulsar-admin brokers healthcheck
T=persistent://public/default/repro
docker exec repro bin/pulsar-admin topics create $T
docker exec repro bin/pulsar-admin topics create-subscription -s sub $T
docker exec repro bin/pulsar-client produce $T -db -m m1,m2
docker exec repro bin/pulsar-client produce $T -db -m m3 -p filter=reject
docker exec repro bin/pulsar-client produce $T -db -m m4
docker exec repro bin/pulsar-client produce $T -db -m m5 -p filter=throw-once
docker exec repro bin/pulsar-client produce $T -db -m m6,m7,m8
docker exec repro bash -c "nohup bin/pulsar-client consume $T -s sub -t Shared -n 0 > /tmp/consumer.log 2>&1 &"
# 20 s later /tmp/consumer.log holds no message; pulsar-admin topics stats $T shows msgBacklog 8
docker exec repro bin/pulsar-client produce $T -db -m m9
# still none delivered; msgBacklog 9
docker exec repro bin/pulsar-admin topics unload $T
# now m1 m2 m4 m5 m6 m7 m8 m9 arrive; msgBacklog 0

Additional information

A possible fix: catch around the per-consumer send so that a throw gives back what no consumer was handed (addMessageToReplay for the read's remaining entries) and reads again after a backoff, settle the filter's entriesToFiltered / entriesToRedeliver lists in a finally, and decrement the pending bytes in a finally. A slice a consumer's sendMessages already took is in that consumer's pending acks and needs nothing more.

Are you willing to submit a PR?

• I'm willing to submit a PR!