[fix][broker] Defer ack state updates until persistence succeeds

[nodece]

Motivation

The broker ack path was updating consumer-side in-memory ack state before the underlying subscription ack path had fully completed. That makes ack failures harder to reason about and can leave broker-side state transitions out of sync with the persisted ack result and ack-receipt flow.

Changes

• Add an async-first subscription ack path and route broker ack handling through it
• Defer consumer unacked and pending-ack state updates until the async ack operation completes successfully
• Keep ack-receipt handling aligned with the async broker ack completion in ServerCnx
• Update subscription implementations that participate in the shared ack flow, including persistent, non-persistent, compactor, replicated-subscription, dispatcher, and pending-ack paths
• Refactor Consumer.individualAckNormal and Consumer.individualAckWithTransaction into a single unified individualAck method, eliminating code duplication between the transactional and non-transactional individual ack paths. The unified method preserves the original timing semantics:
  • Non-transactional acks: pending ack state updates remain deferred until persistence succeeds
  • Transactional acks: pending ack state updates are applied immediately (preserving original behavior), with per-position cleanup scheduled on txn storage completion
• Fix computeAckedCount to return the correct batch size (instead of always 1) when no batch-index ack set is present, ensuring messageAckRate metrics are accurate for batched messages

Known limitation: When a transaction is aborted, the consumer's unacked message count is not restored to its pre-ack value. This is a pre-existing issue unrelated to this refactoring and will be addressed in a follow-up PR.

[codelipenghui]

Will the broker stop dispatch messages to the consumer if the consumer has been set to a lower max unack message and the persistence interval of the cursor set to a higher value?

[nodece]

@codelipenghui

The broker should not stop dispatching messages just because managedLedgerCursorPositionFlushSeconds is higher.

In this implementation, unackedMessages is decremented when acknowledgeMessageAsync(...) completes, and that completion is driven by the managed cursor ack callback path, not by the periodic cursor flush task itself.

For the normal path, the ack immediately triggers async mark-delete/delete persistence, so it does not wait for managedLedgerCursorPositionFlushSeconds.

managedLedgerCursorPositionFlushSeconds is only a fallback for dirty/rate-limited cursor updates. In that case the cursor callback is completed immediately and the actual flush is deferred, so dispatch is not blocked waiting for that interval either.