You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "lhotari (via GitHub)" <gi...@apache.org> on 2026/10/13 05:07:51 UTC
Re: [PR] [fix][client] Serialize chunked-message bookkeeping to fix use-after-free and count/queue drift [pulsar]
lhotari commented on code in PR #26084:
URL: https://github.com/apache/pulsar/pull/26084#discussion_r4254931416
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -1513,28 +1523,76 @@ void messageReceived(CommandMessage cmdMessage, ByteBuf headersAndPayload, Clien
if (isMessageUndecryptable || (numMessages == 1 && !msgMetadata.hasNumMessagesInBatch())) {
if (isChunkedMessage) {
- uncompressedPayload = processMessageChunk(uncompressedPayload, msgMetadata, msgId, messageId, cnx);
- if (uncompressedPayload == null) {
+ ByteBuf compressedAssembledPayload;
+ MessageIdImpl[] chunkedMessageIds = null;
+ // Acks decided under the lock are issued only after it is released, see acknowledgeChunkedMessageIds.
+ List<MessageIdImpl> deferredAcks = new ArrayList<>();
+ // Critical section: append the final chunk, then atomically remove + recycle the ctx, capturing the
+ // assembled (still-compressed) buffer and the chunk ids into locals. Keeping assembly and finalize in
+ // one locked region closes the assemble->finalize race with the expiry/eviction path. Once the ctx is
+ // out of chunkedMessagesMap and recycled, the captured buffer is reachable only through this thread, so
+ // decompression and message building are done below, OUTSIDE the lock, to avoid serializing other
+ // completions and expiry behind codec work (and discardCorruptedMessage on failure).
+ synchronized (chunkedMessageLock) {
+ compressedAssembledPayload =
+ doProcessMessageChunk(uncompressedPayload, msgMetadata, msgId, messageId, cnx,
+ deferredAcks);
+ if (compressedAssembledPayload != null) {
+ // last chunk received: so, stitch chunked-messages and clear up chunkedMsgBuffer
+ log.debug().attr("chunkid", msgMetadata.getChunkId())
+ .attr("totalChunks", msgMetadata.getNumChunksFromMsg())
+ .attr("msgid", msgId)
+ .attr("sequenceid", msgMetadata.getSequenceId())
+ .log("Chunked message completed chunkId, total-chunks, msgId sequenceId");
+
+ // remove the ctx from the map and recycle it, capturing the chunk ids before recycle lets it
+ // be reused. The returned buffer is this ctx's chunkedMsgBuffer; recycle only nulls the field,
+ // the buffer object stays alive and is now owned solely by compressedAssembledPayload.
+ ChunkedMessageCtx chunkedMsgCtx = chunkedMessagesMap.remove(msgMetadata.getUuid());
+ chunkedMessageIds = chunkedMsgCtx.chunkedMessageIds;
+ // Drop the completed uuid from the queue too, mirroring the map removal, so it doesn't linger
+ // as a ghost entry.
+ pendingChunkedMessageUuidQueue.remove(msgMetadata.getUuid());
+ pendingChunkedMessageCount--;
+ chunkedMsgCtx.recycle();
+ }
+ }
+ acknowledgeChunkedMessageIds(deferredAcks);
Review Comment:
**[SUGGESTION] Acks collected under the lock are lost if chunk processing throws**
Moving the acks out of the lock changes one thing: if `doProcessMessageChunk` throws after it has queued ids in `deferredAcks`, those acks never run.
For example, a replacement chunk 0 queues the old context's ids and removes that context. Then `writeBytes` at [`ConsumerImpl.java:1814`](https://github.com/apache/pulsar/blob/3b40ea1e4423bda765b88eb6c05b043310f3c5cf/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L1814) can throw `IndexOutOfBoundsException` when the payload is larger than the advertised `totalChunkMsgSize`. The old chunk ids are no longer in the map or queue, so expiry cannot reach them, and they stay as an ack hole. Before this change they were acked before the replacement. Oldest eviction has the same exposure.
This only happens on a malformed chunk, so it is minor. A cheap fix is to run `acknowledgeChunkedMessageIds(deferredAcks)` in a `finally` after the `synchronized` block, here and in `processMessageChunk`.
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -1513,28 +1521,66 @@ void messageReceived(CommandMessage cmdMessage, ByteBuf headersAndPayload, Clien
if (isMessageUndecryptable || (numMessages == 1 && !msgMetadata.hasNumMessagesInBatch())) {
if (isChunkedMessage) {
- uncompressedPayload = processMessageChunk(uncompressedPayload, msgMetadata, msgId, messageId, cnx);
- if (uncompressedPayload == null) {
- return;
+ ByteBuf compressedAssembledPayload;
+ MessageIdImpl[] chunkedMessageIds;
+ // Critical section: append the final chunk, then atomically remove + recycle the ctx, capturing the
+ // assembled (still-compressed) buffer and the chunk ids into locals. Keeping assembly and finalize in
+ // one locked region closes the assemble->finalize race with the expiry/eviction path. Once the ctx is
+ // out of chunkedMessagesMap and recycled, the captured buffer is reachable only through this thread, so
+ // decompression and message building are done below, OUTSIDE the lock, to avoid serializing other
+ // completions and expiry behind codec work (and discardCorruptedMessage on failure).
+ synchronized (chunkedMessageLock) {
+ compressedAssembledPayload =
+ processMessageChunk(uncompressedPayload, msgMetadata, msgId, messageId, cnx);
+ if (compressedAssembledPayload == null) {
+ return;
+ }
+
+ // last chunk received: so, stitch chunked-messages and clear up chunkedMsgBuffer
+ log.debug().attr("chunkid", msgMetadata.getChunkId())
+ .attr("totalChunks", msgMetadata.getNumChunksFromMsg())
+ .attr("msgid", msgId)
+ .attr("sequenceid", msgMetadata.getSequenceId())
+ .log("Chunked message completed chunkId, total-chunks, msgId sequenceId");
+
+ // remove the ctx from the map and recycle it, capturing the chunk ids before recycle lets it be
+ // reused. The returned buffer is this ctx's chunkedMsgBuffer; recycle only nulls the field, the
+ // buffer object stays alive and is now owned solely by compressedAssembledPayload.
+ ChunkedMessageCtx chunkedMsgCtx = chunkedMessagesMap.remove(msgMetadata.getUuid());
+ chunkedMessageIds = chunkedMsgCtx.chunkedMessageIds;
+ // Drop the completed uuid from the queue too, mirroring the map removal, so it doesn't linger as a
+ // ghost entry.
+ pendingChunkedMessageUuidQueue.remove(msgMetadata.getUuid());
+ pendingChunkedMessageCount--;
+ chunkedMsgCtx.recycle();
}
- // last chunk received: so, stitch chunked-messages and clear up chunkedMsgBuffer
- log.debug().attr("chunkid", msgMetadata.getChunkId())
- .attr("totalChunks", msgMetadata.getNumChunksFromMsg())
- .attr("msgid", msgId)
- .attr("sequenceid", msgMetadata.getSequenceId())
- .log("Chunked message completed chunkId, total-chunks, msgId sequenceId");
-
- // remove buffer from the map, set the chunk message id
- ChunkedMessageCtx chunkedMsgCtx = chunkedMessagesMap.remove(msgMetadata.getUuid());
- if (chunkedMsgCtx.chunkedMessageIds.length > 0) {
- msgId = new ChunkMessageIdImpl(chunkedMsgCtx.chunkedMessageIds[0],
- chunkedMsgCtx.chunkedMessageIds[chunkedMsgCtx.chunkedMessageIds.length - 1]);
+ // Outside the lock: set the chunk message id, decompress the assembled payload, and (only on success)
+ // register it with the unack tracker.
+ if (chunkedMessageIds.length > 0) {
+ msgId = new ChunkMessageIdImpl(chunkedMessageIds[0],
+ chunkedMessageIds[chunkedMessageIds.length - 1]);
}
- // add chunked messageId to unack-message tracker, and reduce pending-chunked-message count
- unAckedChunkedMessageIdSequenceMap.put(msgId, chunkedMsgCtx.chunkedMessageIds);
- pendingChunkedMessageCount--;
- chunkedMsgCtx.recycle();
+ uncompressedPayload =
+ uncompressPayloadIfNeeded(messageId, msgMetadata, compressedAssembledPayload, cnx, false);
Review Comment:
Thanks, this covers it. The release is in a `finally`, and both checked and unchecked codec failures now take the same discard-and-ack path. `testFinalChunkDecompressionFailureThroughMessageReceived` checks both variants. Please do open the follow-up for the `uncompressed` leak in `CompressionCodecZLib.decode`.
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -3182,15 +3258,30 @@ protected void removeExpireIncompleteChunkedMessages() {
if (expireTimeOfIncompleteChunkedMessageMillis <= 0) {
return;
}
- ChunkedMessageCtx chunkedMsgCtx = null;
+ synchronized (chunkedMessageLock) {
Review Comment:
Thanks, this is what I was after. Acks are collected under the lock and issued after it, and the latch-controlled interceptor test shows reception making progress. I left one small note on the call after the `synchronized` block about acks being dropped if processing throws.
##########
pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java:
##########
@@ -115,6 +119,207 @@ public void cleanup() {
}
}
+ /**
+ * Reproduces the chunked-message bookkeeping data race in {@link ConsumerImpl}, using the real
+ * production cross-thread pair:
+ * <ul>
+ * <li>{@code processMessageChunk(...)} on the Netty IO event-loop thread (here: the "receiver" thread);</li>
+ * <li>{@code removeExpireIncompleteChunkedMessages()} on the internalPinnedExecutor (here: the "expirer"
+ * thread).</li>
+ * </ul>
+ *
+ * <p>The faithful scenario is "a late chunk arrives for a uuid that is concurrently being expired": for each uuid
+ * the receiver delivers chunk 0, marks the ctx as already expired (receivedTime = 0), then delivers a late chunk 1
+ * that writes into the same {@code chunkedMsgBuffer} — exactly when the expiry task may release/recycle that ctx.
+ * Without serialization this races (use-after-free / double-recycle). With the fix, both paths take
+ * chunkedMessageLock, so it passes and pendingChunkedMessageCount stays equal to chunkedMessagesMap.size().
+ */
+ @Test(timeOut = 60000)
+ public void testChunkedMessageCountRaceBetweenReceiveAndExpiry() throws Exception {
+ // No max-pending eviction on the receive thread, so only the expiry path removes entries
+ consumerConf.setMaxPendingChunkedMessage(0);
+ createConsumer(consumerConf);
+ // Enable the expiry path (so removeExpireIncompleteChunkedMessages actually works) but skip its lazy
+ // self-scheduling so the test doesn't need a scheduled-executor mock.
+ consumer.expireTimeOfIncompleteChunkedMessageMillis = 1L;
+ consumer.expireChunkMessageTaskScheduled.set(true);
+
+ final int rounds = 5;
+ final int iterations = 50000;
+ final AtomicReference<Throwable> error = new AtomicReference<>();
+
+ for (int round = 0; round < rounds && error.get() == null; round++) {
+ final int currentRound = round;
+ final CountDownLatch start = new CountDownLatch(1);
+ Thread receiver = new Thread(() -> {
+ try {
+ start.await();
+ for (int i = 0; i < iterations && error.get() == null; i++) {
+ String uuid = "uuid-" + currentRound + "-" + i;
+ sendChunk(uuid, 0, 3);
+ ConsumerImpl.ChunkedMessageCtx ctx = consumer.chunkedMessagesMap.get(uuid);
+ if (ctx != null) {
+ // Make it eligible for removeExpireIncompleteChunkedMessages right away.
+ ctx.receivedTime = 0;
+ }
+ // Late chunk for the now-expiring uuid -> writes into the same ctx buffer.
+ sendChunk(uuid, 1, 3);
+ }
+ } catch (Throwable t) {
+ error.compareAndSet(null, t);
+ }
+ }, "receiver-netty-sim");
+
+ Thread expirer = new Thread(() -> {
+ try {
+ start.await();
+ while (receiver.isAlive() || !consumer.chunkedMessagesMap.isEmpty()) {
+ consumer.removeExpireIncompleteChunkedMessages();
+ }
+ } catch (Throwable t) {
+ error.compareAndSet(null, t);
+ }
+ }, "expirer-pinned");
+
+ receiver.start();
+ expirer.start();
+ start.countDown();
+ receiver.join();
+ expirer.join();
+ }
+
+ if (error.get() != null) {
+ throw new AssertionError("chunked-message bookkeeping race: concurrent ctx/buffer access corrupted state",
+ error.get());
+ }
+ int count = consumer.pendingChunkedMessageCount;
+ int mapSize = consumer.chunkedMessagesMap.size();
+ Assert.assertEquals(count, mapSize, "pendingChunkedMessageCount (" + count
+ + ") drifted from chunkedMessagesMap.size() (" + mapSize + ")");
+ }
+
+ private void sendChunk(String uuid, int chunkId, int numChunks) {
+ MessageMetadata md = new MessageMetadata()
+ .setProducerName("p").setSequenceId(0).setPublishTime(System.currentTimeMillis())
+ .setUuid(uuid).setChunkId(chunkId).setNumChunksFromMsg(numChunks).setTotalChunkMsgSize(64);
+ md.setCompression(CompressionType.NONE);
+ MessageIdData idData = new MessageIdData().setLedgerId(1L).setEntryId(chunkId);
+ MessageIdImpl msgId = new MessageIdImpl(1L, chunkId, -1);
+ ByteBuf chunk = Unpooled.wrappedBuffer(new byte[] {1, 2, 3, 4});
+ consumer.processMessageChunk(chunk, md, msgId, idData, null);
+ }
+
+ /**
+ * When a duplicated first chunk (chunkId == 0, redelivered) arrives for a uuid that already has an
+ * in-progress ctx, the old ctx is replaced: removing/recycling it must be paired with decrementing
+ * pendingChunkedMessageCount before the unconditional increment for the new ctx, so the counter stays
+ * equal to the real number of in-progress chunked messages (chunkedMessagesMap.size()).
+ */
+ @Test
+ public void testDuplicateFirstChunkOvercountsPendingChunkedMessageCount() throws Exception {
+ // Disable the lazy expiry self-scheduling so this test doesn't need a scheduled-executor mock.
+ consumer.expireTimeOfIncompleteChunkedMessageMillis = 0L;
+
+ final String uuid = "uuid-dup";
+ // Deliver the first chunk (chunkId == 0) twice for the SAME uuid and SAME (ledgerId, entryId): a redelivered
+ // first chunk of a 2-chunk message. The message never completes, so only the ++/-- bookkeeping is exercised.
+ for (int call = 0; call < 2; call++) {
+ sendChunk(uuid, 0, 2);
+ }
+
+ int count = consumer.pendingChunkedMessageCount;
+ int mapSize = consumer.chunkedMessagesMap.size();
+ Assert.assertEquals(count, mapSize,
+ "a redelivered first chunk over-counted pendingChunkedMessageCount (" + count
+ + ") vs chunkedMessagesMap.size() (" + mapSize + ")");
+
+ // Only this single uuid is ever enqueued, so the queue size equals its occurrence count. After a redelivered
+ // first chunk it must be exactly 1 (stale entry removed, re-added once) — not 2 as the old double-enqueue did.
+ // (GrowableArrayBlockingQueue intentionally doesn't support iteration, so assert on size().)
+ Assert.assertEquals(consumer.pendingChunkedMessageUuidQueue.size(), 1,
+ "uuid should appear exactly once in pendingChunkedMessageUuidQueue but queue size was "
+ + consumer.pendingChunkedMessageUuidQueue.size());
Review Comment:
Thanks, the `finally` with the helper covers this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org