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

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

#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:

// 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:

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

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?

• I'm willing to submit a PR!