[Bug] Compaction horizon and compacted ledger are installed before the commit is durable
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
### Search before reporting
- [X] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.
### Read release policy
- [X] I understand that [unsupported versions](https://pulsar.apache.org/contribute/release-policy/#supported-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:
```java
compactedTopic.newCompactedLedger(position, compactedLedgerId).thenAccept(previousContext -> {
cursor.asyncMarkDelete(position, properties, new MarkDeleteCallback() { ... });
})
```
`PulsarCompactorSubscription.java:88-89`, and `CompactedTopicImpl.newCompactedLedger()` sets the horizon immediately:
```java
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
```text
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?
- [X] I'm willing to submit a PR!
Contributor guide
Assessment
This issue has not been assessed yet.