[Bug] Transaction abort does not restore the consumer's unacked message count
- 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: current `master` (`9ba61bd95de`); line numbers below are `master`.
- The immediate-decrement behavior itself predates #25528 (that PR preserved it deliberately); this issue tracks the follow-up promised in its **Known limitation** note.
### Issue Description
#25528 ("Defer ack state updates until persistence succeeds") unified `individualAckNormal` / `individualAckWithTransaction` into `Consumer.individualAck()`. The transactional branch intentionally keeps the original timing: consumer-level state is applied at ack time, with per-position cleanup on txn storage completion.
The unacked message count is therefore decremented the moment a message is individually acked inside a transaction (`Consumer.java:645-657`):
```java
// TODO: If the transaction is later aborted, the unacked count is NOT restored, leading
// to an incorrect (lower) unacked message count. Fixing this requires coordinating
// with PendingAckHandle's commit/abort callbacks to defer consumer-level state
// updates until the transaction outcome is determined.
if (hasAckSet && ackedCount > 0) {
boolean updated = ackOwnerConsumer.updateRemainingUnacked(...);
if (updated) {
addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount);
}
} else if (!hasAckSet) {
int removed = ackOwnerConsumer.removePendingAckAndGetRemainingUnacked(...);
if (removed != PENDING_ACK_NOT_FOUND) {
addAndGetUnAckedMsgs(ackOwnerConsumer, -removed);
}
}
```
On abort, `PendingAckHandleImpl.internalAbortTxn()` (`PendingAckHandleImpl.java:594-612`) appends the abort mark, clears the per-txn state, and calls `persistentSubscription.redeliverUnacknowledgedMessages(consumer, positions)`. Redelivery goes through the normal dispatch path, which re-registers the positions (`Consumer.java:398`, `addPendingAckIfAllowed`) and re-increments the count (`Consumer.java:440`, `incrementUnackedMessages`).
This makes the impact depend on the ack/abort variant:
1. **Individual acks: window drift.** From the transactional ack until the abort-triggered redelivery is actually delivered, both the consumer-level and subscription-level unacked counts are low. `maxUnackedMessages` flow control under-counts during that window — the consumer is not blocked when it should be and permits are over-issued. The window is the transaction's open duration plus the abort processing latency, which is significant for long-running transactions.
2. **Cumulative transactional acks: permanent drift.** The cumulative abort branch deliberately does not redeliver (`PendingAckHandleImpl.java:580-581`: *"in cumulative ack with transaction, don't depend on server redeliver message, it will cause the messages to be out of order"*), so nothing restores the count for that path.
3. **Consumer gone before redelivery.** If the consumer disconnects before the aborted positions are delivered, the subscription-level count stays low until some consumer eventually consumes those positions.
### Error messages
```text
No error. Nothing is logged at any level; only the in-memory unacked count is wrong.
```
### Reproducing the issue
Analysis is from code. To exercise it deterministically at the integration level:
1. Create a topic and a `Shared` consumer with a modest `maxUnackedMessages` (e.g. via `maxUnackedMessagesOnConsumer` topic policy).
2. Consume N messages, acknowledge them inside a transaction, and check the consumer's unacked count on the broker (metrics / `getUnackedMessages()`): it has dropped by N while the transaction is still open.
3. Keep the transaction open and keep consuming: the consumer is allowed to exceed `maxUnackedMessages` during the window.
4. Abort the transaction: after the redelivery is dispatched the count returns for the individual-ack case; for a cumulative transactional ack it does not return.
A unit test can assert the count directly around `Consumer.individualAck()` with a txn id and a simulated `PendingAckHandleImpl` abort, without waiting for real redelivery.
### Additional information
Suggested direction (as in the in-code TODO): apply consumer-level unacked/pending-ack updates from the `PendingAckHandle` commit/abort callbacks instead of at ack time — the transactional counterpart of what #25528 did for the non-transactional path (non-transactional waits for persistence; transactional should wait for the transaction outcome). The cumulative-abort branch needs an explicit decision for the no-redelivery case.
### Are you willing to submit a PR?
- [X] I'm willing to submit a PR!
Contributor guide
Research direction
Start with Consumer.java:645-657 and PendingAckHandleImpl.java:580-612, then trace the existing pending-ack commit and abort callbacks. Add a unit test around Consumer.individualAck() with a transaction id and simulated abort, covering individual and cumulative acknowledgements. Done means unacked counts remain correct through transaction commit and abort, including the cumulative no-redelivery path.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 56/100