[improve][broker] Add metrics for message position find by timestamp - #26751

Motivation

Seek / reset-cursor by timestamp and message TTL find a position with PersistentMessageFinder, which narrows the search to a range of ledgers using their close timestamps and then runs a binary search over entries (OpFindNewest), reading one entry per probe.

On ledgers offloaded to tiered storage each probe can be expensive: the offload index is sparse (one entry per data block, 64 MiB by default), so reading an entry in the middle of a block can require scanning the block from its start with ranged reads, each one costing tens of milliseconds on an object storage. A seek by timestamp into offloaded data can therefore take much longer than into BookKeeper while the subscription is fenced, and TTL checks on old topics go through the same path.

There is currently no way to observe this: no metric tells how long a find takes, how many entries it reads, or whether they come from BookKeeper or tiered storage. This makes it hard to diagnose slow seeks and to measure improvements to the search.

Modifications

• Add OpenTelemetryMessageFinderStats with the following metrics. Attributes are intentionally limited to low-cardinality values (no topic or subscription), since a broker may own a very large number of topics:
  • pulsar.broker.message.find.duration (histogram, seconds, with bucket boundaries from 1 ms to 60 s), by pulsar.broker.message.find.reason (seek, expiry) and pulsar.broker.message.find.result (found, not_found, failure).
  • pulsar.broker.message.find.entry.read.count (counter, entries) and pulsar.broker.message.find.entry.read.size (counter, bytes), by reason and pulsar.broker.message.find.entry.storage (bookkeeper, offloaded).
• The storage is the one the entry's ledger is actually read from: ManagedLedgerImpl.isReadFromOffloadedLedgerHandle(ledgerId) checks whether the opened read handle is an OffloadedLedgerHandle, so an offloaded ledger still read from BookKeeper (bookkeeper-first read priority) is counted as bookkeeper. The file system offloader's FileStoreBackedReadHandleImpl and BlobStoreBackedReadHandleImplV2 now implement the OffloadedLedgerHandle marker too; their other uses of the marker are unchanged since they keep the default lastAccessTimestamp() of -1, which opts out of the idle handle eviction.
• The entry read metrics count the entries evaluated by the search: entries served from the broker entry cache are included, under the storage of their ledger. This is stated in the metric descriptions.
• PersistentMessageFinder records the metrics for each entry read and when a find completes, and adds the per-find details (duration, entries and bytes read, offloaded entries read, whether the ledger range was narrowed) to the existing Found position closest to provided timestamp log line.
• Wire the stats for the seek path (PersistentSubscription.resetCursor(long)) and the TTL paths (PersistentMessageExpiryMonitor, PersistentTopic). The previous constructor is kept for tests.

Follow-up: document the new metrics in the OpenTelemetry metrics reference on the website (apache/pulsar-site).

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Added PersistentMessageFinderTest.testPersistentMessageFinderMetricsWhenFound (exact entry count and bytes, compared with the reads intercepted in BookKeeper), ...WhenNotFoundForExpiry, ...WhenReadFails (injected read failure) and ...WhenOffloaded.
• Added FileSystemManagedLedgerOffloaderTest.testReadHandleIsOffloadedLedgerHandle and an assertion in BlobStoreManagedLedgerOffloaderStreamingTest.testReadAndWrite on the real read handles.
• Existing tests PersistentMessageFinderTest and SubscriptionSeekTest pass.

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: new pulsar.broker.message.find.* OpenTelemetry metrics, see Modifications
• Anything that affects deployment

🤖 Generated with Claude Code