[Bug] Compaction horizon and compacted ledger are installed before the commit is durable

[lhotari]

Search before reporting

• I searched in the issues and found nothing similar.

Read release policy

• I understand that unsupported versions don't get bug fixes. I reproduced the issue on the current master branch.

User environment

• Broker version: 4.2.4
• Present on master (9ba61bd95de); line numbers below are master.

Issue Description

PulsarCompactorSubscription.acknowledgeMessageAsync() installs the new compacted ledger before it durably records it:

compactedTopic.newCompactedLedger(position, compactedLedgerId).thenAccept(previousContext -> {
    cursor.asyncMarkDelete(position, properties, new MarkDeleteCallback() { ... });
})

PulsarCompactorSubscription.java:88-89, and CompactedTopicImpl.newCompactedLedger() sets the horizon immediately:

compactionHorizon = p;

CompactedTopicImpl.java:74.

If the mark-delete then fails, the broker keeps serving reads from the new compacted ledger at the new horizon while the persisted CompactedTopicLedger property still names the old one. Two consequences:

1. The compacted view can move backwards across a topic reload. In memory the horizon is at the new position; after an unload/reload the subscription is reconstructed from the persisted property (PulsarCompactorSubscription.java:50-58) and reverts to the older ledger and horizon.

2. The two compaction triggers read different notions of progress. PersistentTopic.checkCompaction() (:2512) gates on compactionSub.estimateBacklogSize() > compactionThreshold — cursor-derived. PersistentTopic.triggerCompactionWithCheckHasMoreMessages() (:4873) gates on lastDispatchablePosition.compareTo(lastCompactedPosition) > 0, and lastCompactedPosition is PulsarTopicCompactionService.getLastCompactedPosition() → CompactedTopicImpl.getCompactionHorizon() (PulsarTopicCompactionService.java:107 → CompactedTopicImpl.java:325) — this in-memory value, which the failed run already advanced. So the second gate reflects progress that was never made durable.

The ordering itself is deliberate and the comment above it explains why (PulsarCompactorSubscription.java:81-87): the reader must be able to see the compacted data before the original ledger can be trimmed. The problem is that nothing rolls the in-memory state back when the durable half fails.

Error messages

No error is produced by the divergence itself.

Reproducing the issue

Analysis is from code:

1. Compact a topic so a compacted ledger and horizon exist.
2. Make the subsequent cursor.asyncMarkDelete() fail (e.g. force the cursor into State.NoLedger with no writable bookies).
3. Read PulsarTopicCompactionService.getLastCompactedPosition() and compare it with the CompactedTopicLedger value in the persisted ManagedCursorInfo — they disagree.
4. Unload and reload the topic; the horizon and the served compacted ledger revert.

Additional information

Suggested direction: promote the new ledger and horizon only after the mark-delete is durable — e.g. keep a pending context that markDeleteComplete promotes — or, if the current ordering must be preserved for the reason in the comment, roll the horizon back on markDeleteFailed. Either way the two trigger gates should agree on which value is authoritative.

Related: #25528 made the ack future reflect persistence, so on master a failed mark-delete now fails the compaction run. That reduces the blast radius but does not remove the divergence, because the in-memory horizon has already been advanced by the time the failure is known.

Are you willing to submit a PR?

• I'm willing to submit a PR!