[fix][client] Serialize chunked-message bookkeeping to fix use-after-…

[SongOf]

Motivation

ConsumerImpl's chunked-message reassembly state — the per-uuid ChunkedMessageCtx,
its chunkedMsgBuffer, pendingChunkedMessageCount and pendingChunkedMessageUuidQueue
— is mutated from two different threads with no synchronization:

• the receive/assembly path (processMessageChunk, and the last-chunk finalize in
  messageReceived) runs on the Netty IO event-loop thread
  (ClientCnx.handleMessage calls consumer.messageReceived(...) directly);
• the incomplete-chunk expiry path (removeExpireIncompleteChunkedMessages) runs on
  the client's internalPinnedExecutor — a separate single-thread pool
  (client.getInternalExecutorService()), not the eventLoopGroup.

When a late chunk for a uuid arrives while the expiry task is removing that same ctx,
the expiry thread can release() / recycle() a ChunkedMessageCtx and its buffer while
the receive thread is still writing into it. This races into:

• use-after-free / double-free of chunkedMsgBuffer (IllegalReferenceCountException,
  or worse — writing into memory the allocator already handed to someone else);
• double ChunkedMessageCtx.recycle(), which corrupts the Netty Recycler pool and can
  hand the same instance to two different chunked messages;
• pendingChunkedMessageCount drift (non-atomic int mutated from both threads).

Incomplete-chunk expiry is enabled by default (expireTimeOfIncompleteChunkedMessageMillis
= 1 minute), so any consumer of chunked messages is exposed.

Separately, a redelivered first chunk (chunkId == 0 for a uuid that already has an
in-progress ctx) replaced the old ctx without decrementing pendingChunkedMessageCount
and re-enqueued the uuid, so the counter over-counted and pendingChunkedMessageUuidQueue
ended up with duplicate / mis-ordered entries (the queue is meant to be ordered oldest-first
by receivedTime, with one entry per in-progress uuid, since removeExpire/removeOldest
clean from the head).

Modifications

• Add a dedicated lock chunkedMessageLock. processMessageChunk,
  removeOldestPendingChunkedMessage and removeExpireIncompleteChunkedMessages now acquire
  it (thin wrapper over an extracted body), and messageReceived holds it across the
  last-chunk assembly + finalize so the assemble→finalize window is closed. All chunk
  bookkeeping is serialized, and pendingChunkedMessageCount is mutated only under the lock
  (so it needs no volatile/Atomic). Heavy per-message work (newMessage, decryption of
  non-chunk payloads, executeNotifyCallback) stays outside the lock, and non-chunked
  messages never touch it.
  • The lock only ever guards chunk bookkeeping; the calls made under it
    (ConcurrentHashMap ops, doAcknowledge which is async/non-blocking) never acquire it
    in reverse, so there is no new lock-ordering / deadlock and no blocking call held across
    the lock.
• Fix the redelivered-first-chunk path: decrement pendingChunkedMessageCount for the
  replaced ctx, remove(uuid) its stale entry from pendingChunkedMessageUuidQueue and
  re-add(uuid) at the tail — keeping the queue ordered oldest-first with exactly one entry
  per in-progress uuid (net 0 count change on replace, +1 on a genuinely new uuid).

This is an internal client-side locking change only; no public API, wire protocol, schema,
config defaults or metrics are changed.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• ConsumerImplTest.testChunkedMessageCountRaceBetweenReceiveAndExpiry — drives
  processMessageChunk on a "receiver" thread concurrently with the real
  removeExpireIncompleteChunkedMessages on an "expirer" thread, for the "a late chunk
  arrives for a uuid that is concurrently being expired" scenario. Verified to fail before
  the fix (IllegalReferenceCountException / corrupted pendingChunkedMessageCount) and
  pass after.
• ConsumerImplTest.testDuplicateFirstChunkOvercountsPendingChunkedMessageCount —
  deterministic; delivers a redelivered first chunk and asserts
  pendingChunkedMessageCount == chunkedMessagesMap.size() and that the uuid appears exactly
  once in pendingChunkedMessageUuidQueue. Verified to fail before and pass after.

Full pulsar-client unit suite (712 tests) passes with no regressions.

Does this pull request potentially affect one of the following parts:

• Dependencies (add or upgrade a dependency)
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc-required
• doc-not-needed
  (internal bug fix; no user-facing behavior or config change)
• doc
• doc-complete

[lhotari]

remove this reference

[SongOf]

removed

[lhotari]

referring to findings doesn't make sense in code comments

[SongOf]

removed, thanks for your comments.

[congbobo184]

LGTM

[Denovo1998]

I think this still leaves a stale-queue-head case. The PR now states that pendingChunkedMessageUuidQueue has exactly one entry per in-progress uuid, but not every map removal keeps that invariant. The final-chunk path removes the ctx from chunkedMessagesMap and decrements the count without removing the uuid from pendingChunkedMessageUuidQueue; the unexpected/out-of-order branch also removes/recycles the ctx without removing the queue entry or decrementing pendingChunkedMessageCount.

That matters because doRemoveExpireIncompleteChunkedMessages() uses peek(). If the queue head points to a completed/removed uuid, chunkedMsgCtx is null and the method returns, so later expired incomplete chunked messages behind that stale head are never expired. I reproduced this on the PR head by sending chunk 0 then out-of-order chunk 2 for one uuid, then an expired chunk 0 for another uuid; removeExpireIncompleteChunkedMessages() returns at the stale head and leaves the second ctx in the map.

Could we either remove the uuid from pendingChunkedMessageUuidQueue on every path that removes from chunkedMessagesMap (final completion and unexpected/out-of-order cleanup included), or make the expiry loop drain stale null heads before deciding that the oldest live ctx is not expired?

[SongOf]

fixed in the final-chunk path. and fixed in out-of-order chunk path. At the same time, modify doRemoveExpireIncompleteChunkedMessages to handle null elements at the head of the queue, aligning its logic with the doRemoveOldestPendingChunkedMessage method.

[Denovo1998]

The lock is described as guarding only chunk bookkeeping, but this path currently holds chunkedMessageLock through final-chunk decompression in doProcessMessageChunk() (uncompressPayloadIfNeeded). That can be a relatively expensive codec operation and, on decompression failure, can call discardCorruptedMessage() while the lock is held. This serializes unrelated chunked-message completions and expiry behind decompression work.

Could we narrow the critical section to only append/remove the ctx and capture the assembled buffer / chunk ids, then decompress and build the final message outside the lock? That would still close the assemble -> finalize race without turning the lock into a codec/decompression gate.