[Bug] Mark-delete rate limiter reports an acknowledgement as persisted when nothing was persisted
- 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
#25528 ("Defer ack state updates until persistence succeeds") made an acknowledgement with a receipt mean *"this ack is persisted"*: `ServerCnx.java:2832` passes `hasRequestId` as `requirePersistedAck`, and `Consumer.java:561-563` then composes `subscription.acknowledgeMessageAsync(...)` instead of completing immediately.
The mark-delete rate limiter breaks that guarantee. `ManagedCursorImpl.asyncMarkDelete()` has a throttled path that invokes the success callback **without persisting anything**:
```java
// Apply rate limiting to mark-delete operations
if (markDeleteLimiter != null && !markDeleteLimiter.tryAcquire()) {
isDirty = true;
updateLastMarkDeleteEntryToLatest(newPosition, properties);
callback.markDeleteComplete(ctx);
return;
}
```
`ManagedCursorImpl.java:2330-2334`. `asyncDelete()` has the same shape at `:2702-2706` (`callback.deleteComplete(ctx)`).
Both subscription implementations complete the ack future from that callback:
- `PersistentSubscription.acknowledgeMessageAsync()` → `new AckCallback(previousMarkDeletePosition, future)` (`PersistentSubscription.java:483`, `:508`)
- `PulsarCompactorSubscription.acknowledgeMessageAsync()` → `completionFuture.complete(null)` inside `markDeleteComplete` (`PulsarCompactorSubscription.java:89-99`)
So a throttled mark-delete produces a successful `CommandAckResponse`, and the client is told the position is durable when only in-memory state changed. `flush()` (`ManagedCursorImpl.java:4077`) does re-drive it later, but it clears `isDirty` before calling `asyncMarkDelete()` and is subject to the same limiter, so it is a retry, not a durability barrier.
#### Consequence specific to topic compaction
For the compaction cursor, `markDeleteComplete` is not merely a status report — it is destructive:
```java
public void markDeleteComplete(Object ctx) {
...
if (previousContext != null) {
compactedTopic.deleteCompactedLedger(previousContext.getLedger().getId());
}
completionFuture.complete(null);
}
```
`PulsarCompactorSubscription.java:91-98`. The previous compacted ledger is deleted, while the durably persisted `CompactedTopicLedger` property still names it (the new value has not been written). A broker restart or topic unload inside that window leaves the topic pointing at a deleted ledger, and the compacted view is lost.
This is why the fix matters beyond the reporting inaccuracy: any future work that builds on "the ack future means durable" inherits the same hole.
### Error messages
```text
No error. The ack is answered successfully and nothing is logged at any level.
```
### Reproducing the issue
Analysis is from code. To exercise it deterministically:
1. Set `managedLedgerDefaultMarkDeleteRateLimit` low enough that the limiter throttles (the default of `1.0` combined with Guava `RateLimiter`'s idle permit accumulation means a low-frequency acker rarely trips it).
2. Create a consumer with `ackReceiptEnabled(true)` and acknowledge cumulatively faster than the configured rate.
3. Observe that `CommandAckResponse` reports success for acks that were never persisted — e.g. kill the broker before `flush()` runs and observe the position after recovery.
For the compaction variant, a unit test can call `PulsarCompactorSubscription.acknowledgeMessageAsync()` with a throttling limiter installed and assert that `deleteCompactedLedger` was invoked while the persisted `CompactedTopicLedger` property still holds the old id.
### Additional information
Suggested direction: do not signal completion from a non-persisting path. Either skip the callback and let `flush()` complete it once persisted, or introduce a distinct "accepted, not yet durable" signal so that `requirePersistedAck` callers and `PulsarCompactorSubscription`'s ledger cleanup can distinguish the two.
Note that 4.2.x does not contain #25528, so on that branch the reporting inaccuracy exists for a different reason; the destructive compaction consequence described above applies to both.
### 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.