apache / apache/pulsar

[Bug] Mark-delete rate limiter reports an acknowledgement as persisted when nothing was persisted

Open
#26,484 0 comments 0 reactions 1 assignee Claimed by @lhotari View on GitHub
type/bug
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.