[fix][offload] Cache and reuse entry offsets discovered while skipping in BlobStoreBackedReadHandleImpl

[omarkj]

seekToEntryOffset() falls back to skipPreviousEntry() when it has neither a cached nor a persisted precise offset for the target entry. That walk learns every entry's exact offset along the way but never records it in entryOffsetsCache, and step 2's cache lookup only ever probes for expectedEntryId - 1, so a cached offset is only reachable when consecutive reads land on strictly adjacent entry ids. Real cursor traffic on fragmented acks jumps by the gap between un-acked entries instead, so a cursor whose backlog is fragmented into many individually-acked ranges pays a full rescan of the offload block on every jump across an acked gap.

entryOffsetsCache.put() is now called in skipPreviousEntry() at the point the entry id is confirmed, mirroring the existing call in the main read loop. seekToEntryOffset()'s step 2 becomes a bounded backward probe: walk from expectedEntryId - 1 down to whichever is larger of the sparse index marker's entry id or expectedEntryId - MAX_OFFSET_PROBE (default 1024, pulsar.jclouds.readhandleimpl.offsetprobe.max), seeking and skipping from the first cache hit. A later jump into an already-walked block now resolves in "gap" probes instead of a full rescan, and each walk extends the cached run forward, so only the first jump into a given region pays the full walk cost. Cache growth remains bounded by OffsetsCache's existing maximumSize/expireAfterAccess eviction.

[lhotari]

Real cursor traffic on fragmented acks jumps by the gap between un-acked entries instead, so a cursor whose backlog is fragmented into many individually-acked ranges pays a full rescan of the offload block on every jump across an acked gap.

Not directly related, but there's a slow path case for catch-up reads. Each time BlobStoreBackedReadHandleImpl is opened, it will read and parse the index. The BlobStoreBackedReadHandleImpl instance will get closed by default after 600 seconds of inactivity (managedLedgerInactiveOffloadedLedgerEvictionTimeSeconds=600). The solution added by #19783 isn't optimal. There should also be a separate cache for the read and parsed indexes to avoid the extra work when possible. The cache could be bounded by the estimated memory usage by using Caffeine's maximumWeight/weight feature.

[lhotari]

seekToEntryOffset() falls back to skipPreviousEntry() when it has neither a cached nor a persisted precise offset for the target entry.

I took a closer look at OffsetsCache (originally added in #16417, refactored in #22679) and started to doubt the usefulness of it.
The cache will only avoid calling this.indexEntries.floorEntry(messageEntryId).getValue()

pulsar/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/OffloadIndexBlockImpl.java

Lines 102 to 113 in 01964dc

	@Override
	public OffloadIndexEntry getIndexEntryForEntry(long messageEntryId) throws IOException {
	if (messageEntryId > segmentMetadata.getLastEntryId()) {
	log.warn().attr("entryId", messageEntryId)
	.attr("lastEntryId", segmentMetadata.getLastEntryId())
	.log("Requested entry beyond lastEntryId");
	throw new IndexOutOfBoundsException("Entry index: " + messageEntryId
	+ " beyond lastEntryId: " + segmentMetadata.getLastEntryId());
	}
	// find the greatest mapping Id whose entryId <= messageEntryId
	return this.indexEntries.floorEntry(messageEntryId).getValue();
	}

I wouldn't expect that TreeMap.floorEntry is inefficient. I haven't run a deeper analysis yet.

[lhotari]

I took a closer look at OffsetsCache (originally added in #16417, refactored in #22679) and started to doubt the usefulness of it. The cache will only avoid calling this.indexEntries.floorEntry(messageEntryId).getValue()

Well now I get it. The index contains the offset of the first entry in the uploaded block (blocks are written every 5MB by default (minBlockSizeInBytes)). The role of the OffsetsCache is to be able to store the offset directly to a specific entry without needing to scan each time. Yes, it continues to be useful.

[lhotari]

Thanks! LGTM