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

[nodece]

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: current master (9ba61bd95de); line numbers below are master.
• The immediate-decrement behavior itself predates [fix][broker] Defer ack state updates until persistence succeeds #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):

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

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?

• I'm willing to submit a PR!