[fix][broker] Prevent automatic TTL expiry from skipping reader messages - #26505

Fixes #26504

Motivation

A Reader attaches to a topic through a non-durable subscription, and the cursor of that subscription is the reader's scan position — a reader has no backlog and no acknowledgements.

The scheduled TTL task, PersistentTopic.checkMessageExpiry(), walks every subscription of the topic and force-acknowledges everything older than the TTL. It excludes the compaction cursor and the configured additionalSystemCursorNames, but it does not exclude non-durable cursors, so a reader's cursor gets expired like any consumer backlog.

For a durable subscription that is exactly what TTL is defined to do. For a reader it silently drops data, because ManagedCursorImpl.setAcknowledgedPosition() drags the read position along whenever the new mark-delete position reaches or passes it:

if (currentReadPosition.compareTo(markDeletePosition) <= 0) {
    // If the position that is mark-deleted is past the read position, it
    // means that the client has skipped some entries. We need to move
    // read position forward
    Position newReadPosition = ledger.getNextValidPosition(markDeletePosition);

A reader that is still scanning messages older than the TTL is therefore teleported past everything the expiry task just acknowledged. It gets no error and no warning — it simply resumes after the gap and returns fewer messages.

That is the behaviour reported in #26504 (from discussion #26501): a reader started at MessageId.earliest on a topic with ~10 year retention and a 2 week TTL returned anywhere between 2,647 and 24,756 of the same ~24.7k messages across ten consecutive runs. The outcome depends only on whether the periodic expiry task (messageExpiryCheckIntervalInMinutes, 5 minutes by default) fires while the reader is still behind the TTL boundary; raising the TTL past the age of the oldest message made the problem disappear. No message was ever deleted — retention covered them all — the reader was just moved past them.

Retention, not TTL, is what bounds how long a reader may keep reading. When ManagedLedgerImpl actually trims ledgers, advanceCursorsIfNecessary() force-advances non-durable cursors over the deleted data, so excluding them from TTL expiry cannot let a stale reader pin storage.

Modifications

PersistentTopic: add a sub.getCursor().isDurable() condition to the subscription filter in both automatic expiry paths — checkMessageExpiryWithSharedPosition() (the ManagedLedgerImpl path) and checkMessageExpiryWithoutSharedPosition() (the fallback for custom ManagedLedger implementations).

Durable subscriptions and replicators are unaffected, and only the automatic TTL task changes. The administrative expiry endpoints (pulsar-admin topics expire-messages and expire-messages-all-subscriptions) iterate subscriptions in PersistentTopicsBase and still act on reader subscriptions, so an operator can still expire them explicitly.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• New ReaderMessageTTLTest.testReaderRetainsUnreadMessagesDuringExpiry publishes 20 messages to a topic with infinite retention and a 1 second TTL, reads the first message, pauses the reader until every message is older than the TTL, then drives checkMessageExpiry() until the durable subscription's backlog reaches zero. It asserts that the reader's mark-delete position never passes its read position and that the remaining 19 messages are all still readable, in order.
• The test runs over both expiry paths via a @DataProvider: the shared-position path, and — with a ManagedLedger mock that delegates to the real one — the per-subscription fallback path.
• Both data provider cases fail on master (AssertionError: [TTL must not acknowledge unread reader messages]) and pass with the fix.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment