[fix][client] Avoid ack hole for chunk message by liangyepianzhou · Pull Request #21101 · apache/pulsar

liangyepianzhou · 2023-08-31T06:07:33Z

Motivation

Handle ack hole case:
For example:

                     Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1
                     Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2
                     Chunk-3 sequence ID: 0, chunk ID: 0, msgID: 1:3
                     Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:4
                     Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:5

Consumer ack chunk message via ChunkMessageIdImpl that consists of all the chunks in this chunk
message(Chunk-3, Chunk-4, Chunk-5). The Chunk-1 and Chunk-2 are not included in the
ChunkMessageIdImpl, so we should process it here.

Modification

Ack chunk-1 and chunk-2.

Verifying this change

Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

Added integration tests for end-to-end deployment with large payloads (10MB)
Extended integration test for recovery after broker failure

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

If the box was checked, please highlight the changes

Documentation

doc
doc-required
doc-not-needed
doc-complete

Matching PR in forked repository

PR in forked repository:

## Motivation Handle ack hole case: For example: Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1 Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2 Chunk-3 sequence ID: 0, chunk ID: 0, msgID: 1:3 Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:4 Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:5 Consumer ack chunk message via ChunkMessageIdImpl that is consist of all the chunks in this chunk message(Chunk-3, Chunk-4, Chunk-5). The Chunk-1 and Chunk-2 are not included in the ChunkMessageIdImpl, so we should process here. ## Modification Ack chunk-1 and chunk-2.

Technoboy- · 2023-08-31T09:14:06Z

Is it related to this flaky test ?
https://github.com/apache/pulsar/actions/runs/5934883964/job/16092673164?pr=21043

liangyepianzhou · 2023-08-31T09:18:29Z

Is it related to this flaky test ? https://github.com/apache/pulsar/actions/runs/5934883964/job/16092673164?pr=21043

No, It should not be related to this flaky test. IMO, this test may be errored.

liangyepianzhou · 2023-08-31T09:43:33Z

The root cause may be doAcknowledge(msgId, AckType.Individual, Collections.emptyMap(), null); it is an async method. I will open a PR to fix that flaky test.

codecov-commenter · 2023-08-31T09:53:52Z

Codecov Report

❌ Patch coverage is 0% with 13 lines in your changes missing coverage. Please review.
✅ Project coverage is 34.61%. Comparing base (dab5b2f) to head (b45ca0a).
⚠️ Report is 2202 commits behind head on master.

Files with missing lines	Patch %	Lines
...va/org/apache/pulsar/client/impl/ConsumerImpl.java	0.00%	13 Missing ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #21101       +/-   ##
=============================================
- Coverage     73.03%   34.61%   -38.42%     
+ Complexity    32363    12044    -20319     
=============================================
  Files          1887     1698      -189     
  Lines        139860   130285     -9575     
  Branches      15384    14238     -1146     
=============================================
- Hits         102140    45098    -57042     
- Misses        29634    79187    +49553     
+ Partials       8086     6000     -2086

Flag	Coverage Δ
inttests	`∅ <ø> (?)`
systests	`25.04% <0.00%> (-0.07%)`	⬇️
unittests	`31.85% <0.00%> (-40.56%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...va/org/apache/pulsar/client/impl/ConsumerImpl.java	`49.40% <0.00%> (-27.15%)`	⬇️

... and 1468 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

## Motivation Handle ack hole case: For example: ```markdown Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1 Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2 Chunk-3 sequence ID: 0, chunk ID: 0, msgID: 1:3 Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:4 Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:5 ``` Consumer ack chunk message via ChunkMessageIdImpl that consists of all the chunks in this chunk message(Chunk-3, Chunk-4, Chunk-5). The Chunk-1 and Chunk-2 are not included in the ChunkMessageIdImpl, so we should process it here. ## Modification Ack chunk-1 and chunk-2. (cherry picked from commit 59a8e72)

## Motivation Handle ack hole case: For example: ```markdown Chunk-1 sequence ID: 0, chunk ID: 0, msgID: 1:1 Chunk-2 sequence ID: 0, chunk ID: 1, msgID: 1:2 Chunk-3 sequence ID: 0, chunk ID: 0, msgID: 1:3 Chunk-4 sequence ID: 0, chunk ID: 1, msgID: 1:4 Chunk-5 sequence ID: 0, chunk ID: 2, msgID: 1:5 ``` Consumer ack chunk message via ChunkMessageIdImpl that consists of all the chunks in this chunk message(Chunk-3, Chunk-4, Chunk-5). The Chunk-1 and Chunk-2 are not included in the ChunkMessageIdImpl, so we should process it here. ## Modification Ack chunk-1 and chunk-2.

…ages (#1464) Master Issue: #1446 related issue apache/pulsar#21070 and apache/pulsar#21101 ### Motivation Current, when the producer resend the chunked message like this: ``` M1: UUID: 0, ChunkID: 0 M2: UUID: 0, ChunkID: 0 // Resend the first chunk M3: UUID: 0, ChunkID: 1 ``` When the consumer received the M2, it will find that it's already tracking the UUID:0 chunked messages, and will then discard the message M1 and M2. This will lead to unable to consume the whole chunked message even though it's already persisted in the Pulsar topic. Here is the code logic: ```Go if ctx == nil || ctx.chunkedMsgBuffer == nil || chunkID != ctx.lastChunkedMsgID+1 { lastChunkedMsgID := -1 totalChunks := -1 if ctx != nil { lastChunkedMsgID = int(ctx.lastChunkedMsgID) totalChunks = int(ctx.totalChunks) ctx.chunkedMsgBuffer.Clear() } pc.log.Warnf(fmt.Sprintf( "Received unexpected chunk messageId %s, last-chunk-id %d, chunkId = %d, total-chunks %d", msgID.String(), lastChunkedMsgID, chunkID, totalChunks)) pc.chunkedMsgCtxMap.remove(uuid) pc.availablePermits.inc() return nil } ``` The bug can be easily reproduced using the testcase `TestChunkWithReconnection` and `TestResendChunkMessages` introduced by this PR. ### Modifications The current chunk processing strategy is consistent with the behavior of the Java client: https://github.com/apache/pulsar/blob/52a4d5ee84fad6af2736376a6fcdd1bc41e7c52f/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L1579 When receiving the new duplicated first chunk of a chunked message, the consumer discard the current chunked message context and create a new context to track the following messages. For the case mentioned in Motivation, the M1 will be released and the consumer will assemble M2 and M3 as the chunked message.

…-order chunks (#587) Master Issue: apache/pulsar#13627 Related Issue: apache/pulsar#21070 and apache/pulsar#21101 ### Motivation apache/pulsar#21070 and apache/pulsar#21101 fixed two critical issues in the Java client's chunked message handling: 1. **Unable to reassemble chunked messages after redeliver**: When a chunked message is redelivered (e.g., due to broker unload or reconnect), the consumer receives duplicated chunks. The old code could not handle this correctly: - For duplicated first chunk (chunkId=0): the old context was not properly cleaned up and restarted, causing the message to never be assembled. - For duplicated middle chunks: the chunk would be rejected (since chunkId ≤ lastChunkedMessageId), and the old code would discard the context entirely, making the message unrecoverable. 2. **Ack holes caused by corrupted or orphaned chunks**: When a different producer reuses the same uuid (corrupted chunk scenario), or when chunk context is discarded due to gap/expiration, the stale cached chunks or the incoming corrupted chunks were never acknowledged. This causes the broker subscription cursor to get stuck, leading to message backlog accumulation that never drains — even after all logically valid messages have been consumed and acknowledged. These PRs added logic to distinguish between redeliver (same messageId) and corruption (different messageId), allowing the consumer to correctly restart chunk assembly on redeliver while acking stale chunks on corruption to prevent ack holes. The C++ client had the same issues. This PR ports the equivalent logic to ensure consistent behavior across all client implementations. **Note**: Currently, after a chunked message is assembled, the ackTimeout and nack logic only tracks/handles the last chunk message (i.e., the final messageId of the assembled message). This means if ackTimeout or nack triggers a redeliver, only the last chunk entry is redelivered rather than all chunk entries. This limitation needs to be addressed in a follow-up PR. ### Modifications **Core logic changes in `ConsumerImpl.cc` (`processMessageChunk`)**: - **Part 1 (chunkId == 0)**: When receiving a duplicated first chunk for a uuid that already has an incomplete context, detect whether it's a redeliver (same messageId in cache) or corruption (different messageId). For redeliver: remove old context and restart assembling. For corruption: ack all cached chunks to avoid ack holes, then restart. - **Part 3 (duplicated middle chunk)**: When receiving a chunk with chunkId ≤ lastChunkedMessageId, detect whether it's a redeliver or corruption. For redeliver: simply discard the duplicate and continue waiting for the next expected chunk. For corruption: ack the corrupted chunk to avoid ack holes. - **Part 3 (gap chunk)**: When receiving a chunk that skips expected sequence numbers, ack the chunk if it has expired to avoid ack holes. - **Removed `trackMessage` calls for discarded chunks**: The old code called `trackMessage(messageId)` for orphaned/invalid chunks (Part 2 and old Part 3), which would add the single chunk entry to the `UnAckedMessageTracker`. When ackTimeout triggered, it would redeliver only that single chunk entry — but this is pointless because the consumer still cannot assemble a complete chunked message from a single chunk, and the redelivered chunk would just enter the same discard path again in an infinite loop. - Added `LOG_WARN` and `LOG_INFO` for observability across all scenarios. - Added detailed comments explaining each part of the chunk processing logic with examples. **Test changes in `MessageChunkingTest.cc`**: - Added `testResendChunkMessagesWithoutAckHole`: Verifies that resending the first chunk (chunkId=0) allows correct reassembly without ack holes. - Added `testResendChunkMessages`: Verifies interleaved chunk resends across multiple uuids assemble correctly. - Added `testResendChunkWithAckHoleMessages`: Verifies duplicated middle chunks are filtered correctly and chunk gaps cause context cleanup. - Refactored existing tests to reuse the `sendSingleChunk` helper function for better readability.

github-actions Bot added the doc-not-needed Your PR changes do not impact docs label Aug 31, 2023

liangyepianzhou marked this pull request as draft August 31, 2023 06:44

add test

dbd7636

liangyepianzhou marked this pull request as ready for review August 31, 2023 07:04

liangyepianzhou added the ready-to-test label Aug 31, 2023

liangyepianzhou closed this Aug 31, 2023

liangyepianzhou reopened this Aug 31, 2023

liangyepianzhou added release/2.9.5 release/3.0.2 release/2.10.6 release/3.1.1 labels Aug 31, 2023

Technoboy- added this to the 3.2.0 milestone Aug 31, 2023

Technoboy- assigned liangyepianzhou Aug 31, 2023

codelipenghui reviewed Aug 31, 2023

View reviewed changes

Comment thread pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

Comment thread pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java Outdated

Comment thread pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java

optimize

f6595f8

poorbarcode requested review from BewareMyPower, RobertIndie and poorbarcode September 1, 2023 03:38

codelipenghui reviewed Sep 1, 2023

View reviewed changes

optimize

48a7ee0

codelipenghui reviewed Sep 1, 2023

View reviewed changes

Comment thread pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java Outdated

optimize

b45ca0a

codelipenghui approved these changes Sep 1, 2023

View reviewed changes

BewareMyPower approved these changes Sep 1, 2023

View reviewed changes

poorbarcode approved these changes Sep 1, 2023

View reviewed changes

liangyepianzhou merged commit 59a8e72 into apache:master Sep 4, 2023

liangyepianzhou added the cherry-picked/branch-2.10 label Sep 4, 2023

liangyepianzhou added the cherry-picked/branch-2.11 label Sep 4, 2023

liangyepianzhou deleted the ack-hole branch September 4, 2023 01:41

Technoboy- added the cherry-picked/branch-3.1 label Sep 5, 2023

Technoboy- added the cherry-picked/branch-3.0 label Sep 5, 2023

RobertIndie mentioned this pull request Jul 29, 2025

Unexpected chunk messages are not redelivered apache/pulsar-client-go#1397

Open

geniusjoe mentioned this pull request Jan 29, 2026

[Issue 1446][Consumer] Fix consumer can't consume resent chunked messages apache/pulsar-client-go#1464

Merged

1 task

geniusjoe mentioned this pull request Jun 1, 2026

[improve][client-cpp] Refactor processMessageChunk to handle ack holes and out-of-order chunks apache/pulsar-client-cpp#587

Merged

1 task

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[fix][client] Avoid ack hole for chunk message - #21101

[fix][client] Avoid ack hole for chunk message#21101
liangyepianzhou merged 5 commits into
apache:masterfrom
liangyepianzhou:ack-hole

liangyepianzhou commented Aug 31, 2023

Uh oh!

Technoboy- commented Aug 31, 2023

Uh oh!

liangyepianzhou commented Aug 31, 2023 •

edited

Loading

Uh oh!

liangyepianzhou commented Aug 31, 2023 •

edited

Loading

Uh oh!

codecov-commenter commented Aug 31, 2023 •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

Uh oh!

Conversation

liangyepianzhou commented Aug 31, 2023

Motivation

Modification

Verifying this change

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

Documentation

Matching PR in forked repository

Uh oh!

Technoboy- commented Aug 31, 2023

Uh oh!

liangyepianzhou commented Aug 31, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

liangyepianzhou commented Aug 31, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

codecov-commenter commented Aug 31, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

liangyepianzhou commented Aug 31, 2023 •

edited

Loading

liangyepianzhou commented Aug 31, 2023 •

edited

Loading

codecov-commenter commented Aug 31, 2023 •

edited

Loading