[fix][broker] Fix readCompacted skipping tombstones within compaction horizon

[grishaf]

Summary

readCompacted=true readers could miss delete tombstones when a second compaction ran before those tombstones were consumed. CompactedTopicUtils jumped the cursor to horizon + 1 whenever the compacted ledger returned no entry, skipping unread messages in the original topic between the current position and the compaction horizon.

The fix falls back to the original managed ledger at the current cursor instead of seeking past the horizon.

Issue

readCompacted uses a hybrid read path:

• positions ≤ compaction horizon → compacted BookKeeper ledger
• positions > horizon → original managed ledger

When findStartPoint returns NEWER_THAN_COMPACTED (e.g. a tombstone exists in the original topic but not in the compacted ledger), readCompactedEntries returns empty. The broker then seeked to horizon + 1, skipping tombstones still sitting in the original ledger between the reader cursor and the horizon.

Race

1. Reader opens with readCompacted=true and consumes the compacted snapshot.
2. Producer sends tombstones (key-only deletes).
3. Compaction runs again before the continuous reader consumes those tombstones.
4. Reader asks for the next message; compacted ledger has no entry at the current position.
5. Bug: broker seeks to horizon + 1 → tombstones are never delivered.

Impact on TableView

TableView builds its map from a continuous readCompacted reader. If tombstones are skipped:

• deleted keys stay in the in-memory map
• onUpdate/onRemove callbacks for those deletes never fire
• consumers see stale entries that should have been removed

Any client using a long-lived readCompacted reader or TableView (Java, Go, etc.) inherits this broker behavior; no client-side change is required once the broker stops skipping.

Fix

In CompactedTopicUtils, when the compacted read returns empty, read from the original managed ledger at the current cursor (readEntries / readEntriesWithSkipOrWait) instead of seeking to horizon + 1.

Made with Cursor

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[Denovo1998]

An empty compacted read is also the normal end-of-snapshot signal, and readCompactedEntries returns an empty list for COMPACT_LEDGER_EMPTY, NEWER_THAN_COMPACTED, and NoSuchElementException. It does not necessarily mean that this consumer has already transitioned to the original-ledger tail.

Falling back to readEntries here makes a newly created readCompacted reader replay entries that compaction intentionally removed. The modified test demonstrates this: after reading keys 0–4 from the compacted ledger, the reader now receives the pre-delete values and tombstones for keys 5–9. This contradicts the ReaderBuilder.readCompacted contract, which says that only the latest value for each key should be exposed up to the compaction horizon.

Could we preserve the initial compacted-snapshot behavior and track an explicit snapshot/tail phase for the dispatcher? Once a reader has completed its initial snapshot, it should remain on the original ledger even when a later compaction advances the global horizon.

[Denovo1998]

This changes the expected readCompacted semantics rather than only adding coverage for the continuous-reader race.

A fresh reader created after compaction should still see only keys 0–4. Accepting the pre-delete values and tombstones from the original ledger masks the regression introduced by the production change.

Please keep the original fresh-reader expectation and add a separate test where the reader has already completed the initial compacted snapshot before the tombstones are published and the second compaction runs.

Also, this loop only checks the key range. It does not verify the exact number of messages, ordering, duplicate delivery, or whether the messages are actually tombstones.

[Denovo1998]

The test name and Javadoc describe a TableView regression, but the test only creates a Reader.

Could this either be renamed to describe the continuous readCompacted reader behavior, or use an actual TableView and verify that the deleted keys are removed from the map and that listeners receive the null-value updates? That would directly cover the user-visible impact described in the PR.

[lhotari]

Combined local review — Claude Code (Fable 5) + Codex (gpt-5.6-sol)

This comment merges the findings of two independent local AI-assisted reviews (Claude Code running Claude Fable 5, and Codex running gpt-5.6-sol), reviewed and submitted with human oversight.

Overall assessment: The fix is correct and the direction is sound. The old empty compacted read → seek(horizon + 1) behavior applied compaction retroactively to a reader's unread range, silently dropping everything in (last surviving compacted entry, horizon] — including tombstones, which is what breaks TableView delete propagation. Falling back to the original managed ledger restores the "reader raced ahead of compaction" view. Both reviews independently confirmed:

• No duplicate or out-of-order delivery. The compacted read returns empty only when no surviving compacted entry exists at or after the cursor position, so every raw entry in the fallback window is a tombstone, a value later superseded by a tombstone, or a compaction-dropped null-key message — none of which the reader has seen — and they arrive in position order, so per-key state converges.
• No dispatcher hang or busy-spin. The only production caller is PersistentDispatcherSingleActiveConsumer (always wait=true), and in the fallback the cursor position is ≤ horizon ≤ LAC, so readEntriesWithSkipOrWait reads immediately rather than parking.
• The removed readPosition > seekToPosition guard was dead code (the outer branch already guarantees readPosition ≤ horizon).

---

Findings

1. Observable readCompacted semantics change beyond tombstones — should be stated explicitly (CompactedTopicUtils.java, the empty-entries fallback)

An empty compacted read does not identify tombstones specifically. The fallback therefore delivers all raw messages in the tail window:

• superseded values arrive before their tombstone, so a TableView can transiently resurrect a deleted key before removing it (converges, but observable);
• with the default topicCompactionRetainNullKey=false, null-key messages that compaction dropped are now delivered in this window — while null-key messages falling between surviving compacted entries are still skipped, an inconsistency inherent to the approach;
• when compaction produced an empty compacted ledger, the reader now receives the entire retained raw history instead of nothing.

The updated testHasMessageAvailableWithNullValueMessage codifies this (the reader now receives keys 5–9's pre-delete values plus tombstones, where before it received nothing). The two reviews weighed this differently — Codex rates it HIGH (strict readCompacted semantics are weakened); Claude Code considers it an acceptable, defensible trade-off (a reader positioned before those messages would have seen them anyway had it read before compaction ran, and there is no way to deliver tombstones without also exposing this window short of broker-side payload inspection). Either way, the PR description currently only mentions tombstones — please state the full semantics change explicitly so maintainers can make the call consciously.

2. Residual limitation: tombstones already trimmed by retention cannot be recovered (CompactedTopicUtils.java)

The fallback depends on the tombstones still being present in the original managed ledger. Once the compaction cursor has advanced and retention trims those ledgers, no read-path fix can deliver them — the managed ledger fast-forwards reads past deleted ledgers (ManagedLedgerImpl.getNextValidLedger), so the trimmed case degrades to the old skip behavior (no hang or failure). This PR therefore narrows the TableView delete-loss window rather than fully closing it; with ordinary retention the race remains for tombstones whose source ledger was already deleted. The new integration test uses infinite retention (CompactionTest.java), which is fine for reproducing the bug, but the residual limitation is worth documenting, and a normal-retention test would pin down the degraded behavior explicitly.

3. Behavioral contract change for pluggable TopicCompactionService implementations (TopicCompactionService.java / CompactedTopicUtils.java)

TopicCompactionService is a public, pluggable interface (PIP-278), and readCompactedEntries may return an empty list without distinguishing "cursor is in a tombstone gap", "compacted ledger is empty", or a plugin-specific condition. The utility now interprets every empty result as authorization to bypass the compaction service and read raw history within the horizon — a third-party implementation that returned empty to deliberately hide data will now leak raw messages. At minimum the readCompactedEntries javadoc should document the new expectation; a richer result type that disambiguates the empty cases would be the more robust long-term fix.

4. Test coverage and naming (CompactedTopicTest.java, CompactionTest.java)

• testTableViewMissesDeletesWhenCompactionRunsBeforeTombstonesConsumed names the bug rather than the expected behavior, and despite the name it uses a raw readCompacted reader, not a TableView — consider renaming (e.g. testReaderReceivesTombstonesWhenCompactionRunsBeforeConsumption) or adding a TableView-based assertion (onUpdate/removal callbacks) to cover the headline scenario directly.
• The new test does not assert that no additional messages arrive after the three expected tombstones — a trailing assertNull(reader.readNext(1, TimeUnit.SECONDS)) would tighten it.
• Note that with receiverQueueSize(1) the first tombstone is typically prefetched into the client queue before the second compaction runs, so only the later tombstones exercise the fixed path — still sufficient, since the old code deterministically fails on those.
• The rewritten CompactedTopicUtilsTest properly asserts no seek plus the raw-read fallback. Additional coverage worth considering: empty-compacted-ledger, normal-retention, transaction maxReadPosition, and first-read-from-earliest cases.

5. Process nit

The PR body doesn't follow the apache/pulsar PR template (Motivation / Modifications / Documentation checkbox), which the CI bot will flag, and there is no linked issue for the underlying bug.

---

Neither review found duplicate delivery, out-of-order delivery, dispatcher hangs, or busy-spin regressions — the core change is a real fix for a real data-visibility bug, with the semantics/contract implications above to resolve in description, javadoc, and tests.