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

["lhotari (via GitHub)" <gi...@apache.org>]

lhotari opened a new issue, #26484:
URL: https://github.com/apache/pulsar/issues/26484

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


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org