apache / apache/pulsar

[Bug] Transaction abort does not restore the consumer's unacked message count

Open
#26,489 0 comments 1 reaction 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.