Search before reporting
Read release policy
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:
- 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).
- Create a consumer with
ackReceiptEnabled(true) and acknowledge cumulatively faster than the configured rate.
- 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?
Search before reporting
Read release policy
masterbranch.User environment
master(9ba61bd95de); line numbers below aremaster.Issue Description
#25528 ("Defer ack state updates until persistence succeeds") made an acknowledgement with a receipt mean "this ack is persisted":
ServerCnx.java:2832passeshasRequestIdasrequirePersistedAck, andConsumer.java:561-563then composessubscription.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: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)insidemarkDeleteComplete(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 clearsisDirtybefore callingasyncMarkDelete()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,
markDeleteCompleteis not merely a status report — it is destructive:PulsarCompactorSubscription.java:91-98. The previous compacted ledger is deleted, while the durably persistedCompactedTopicLedgerproperty 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
Reproducing the issue
Analysis is from code. To exercise it deterministically:
managedLedgerDefaultMarkDeleteRateLimitlow enough that the limiter throttles (the default of1.0combined with GuavaRateLimiter's idle permit accumulation means a low-frequency acker rarely trips it).ackReceiptEnabled(true)and acknowledge cumulatively faster than the configured rate.CommandAckResponsereports success for acks that were never persisted — e.g. kill the broker beforeflush()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 thatdeleteCompactedLedgerwas invoked while the persistedCompactedTopicLedgerproperty 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 thatrequirePersistedAckcallers andPulsarCompactorSubscription'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?