[fix][broker]Fixed an issue where the entire subscription would be blocked when a chunk message with an ID of zero did not exist. #25120

zjxxzjwang · 2025-12-30T08:39:41Z

Motivation

1、A message is split into chunk0, chunk1, and chunk2. When chunk0 is unexpectedly lost, the current code logic will endlessly deliver chunk1 and chunk2 to redeliveryMessages, causing the entire subscription to become blocked.

2、When a chunk message corresponds to a producer whose availablePermits remain below zero for an extended period, that chunk message will also be endlessly delivered to the “redeliveryMessages” queue, similarly causing the entire subscription to become blocked.

Modifications

1、In the event of chunk0 being unexpectedly lost, chunk1 and chunk2 messages are directly acknowledged and logged without blocking the entire subscription.

2、When sending each chunk of the same message, the availablePermits of the consumer corresponding to that chunk message are no longer checked, ensuring normal delivery.

Documentation

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

…ocked when a chunk message with an ID of zero did not exist.

github-actions · 2025-12-30T08:40:09Z

@zjxxzjwang Please add the following content to your PR description and select a checkbox:

- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

Copilot

Pull request overview

This PR fixes a critical issue where subscriptions become blocked when chunk messages have problems: either when chunk0 is missing, or when a consumer's available permits remain negative. The fix acknowledges and logs orphaned non-first chunks instead of endlessly retrying them, and removes permit checking for subsequent chunks after the first chunk to prevent permit-related blocking.

Key changes:

Modified chunk message handling to directly acknowledge non-first chunks when chunk0 is missing
Removed permit validation for chunk messages after the first chunk to prevent blocking when permits are exhausted
Added subscription parameter to SharedConsumerAssignor to enable acknowledgment of orphaned chunks

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 7 comments.

File	Description
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SharedConsumerAssignor.java	Core fix: inline chunk handling logic, add subscription parameter for acknowledgment, remove getConsumerForUuid method, and handle missing chunk0 scenario
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java	Pass subscription to SharedConsumerAssignor constructor
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumersClassic.java	Pass subscription to SharedConsumerAssignor constructor
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SharedConsumerAssignorTest.java	Update test to accommodate new constructor signature with null subscription

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SharedConsumerAssignorTest.java

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SharedConsumerAssignor.java

lhotari · 2026-01-02T11:13:40Z

@zjxxzjwang please resolve the merge conflict

zjxxzjwang · 2026-01-20T09:25:06Z

@zjxxzjwang please resolve the merge conflict

ok

# Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SharedConsumerAssignor.java

…ithin a cycle

zjxxzjwang · 2026-01-20T10:11:11Z

@lhotari Hello, please review this PR. It fixes a critical bug related to the existence of chunk messages.

Copilot

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SharedConsumerAssignor.java

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SharedConsumerAssignorTest.java

…ocked when a chunk message with an ID of zero did not exist.

lhotari

Please check the review comments. I think that this behavior should be active only when autoSkipNonRecoverableData is enabled in broker.conf. That would be a safe option for avoiding unintentional data loss.

lhotari · 2026-02-04T17:41:47Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SharedConsumerAssignor.java

+        consumerToEntries.computeIfAbsent(consumer, __ -> new ArrayList<>()).add(entry);
+    }
+
+    private boolean skipChunk(EntryAndMetadata entryAndMetadata, MessageMetadata metadata) {


skipChunk sounds like a command, it's better to rename it to make it sound like a query.

Suggested change

private boolean skipChunk(EntryAndMetadata entryAndMetadata, MessageMetadata metadata) {

private boolean shouldSkipChunk(EntryAndMetadata entryAndMetadata, MessageMetadata metadata) {

lhotari · 2026-02-04T17:50:29Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SharedConsumerAssignor.java

+            if (subscription != null) {
+                log.warn("[{}][{}] Skip the message because it is not the first chunk."
+                                + " Position: {}, UUID: {}, ChunkId: {}, NumChunksFromMsg: {}",
+                        subscription.getTopicName(), subscription.getName(), entryAndMetadata.getPosition(),
+                        metadata.getUuid(), metadata.getChunkId(), metadata.getNumChunksFromMsg());
+                // Directly ack the message.
+                if (!(subscription instanceof PulsarCompactorSubscription)) {
+                    subscription.acknowledgeMessage(Collections.singletonList(
+                            entryAndMetadata.getPosition()), AckType.Individual, Collections.emptyMap());
+                }
+            }
+            entryAndMetadata.release();
+            return true;


Since there's a potential for data loss, acknowledging the messages should be active only if autoSkipNonRecoverableData is set in broker.conf. The log message should be logged with ERROR level when autoSkipNonRecoverableData isn't set and the message shouldn't get acknowledged. I think it's fine to skip the message in that case so that processing the subscription continues, but there will be a backlog left behind due to the unacked messages.

lhotari · 2026-02-04T17:54:20Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SharedConsumerAssignor.java

+        if (consumerForUuid == null) {
+            if (skipChunk(entryAndMetadata, metadata)) {
+                return availablePermits;
+            }
+            consumerForUuid = consumer;
+            uuidToConsumer.put(uuid, consumerForUuid);
+        }


It seems that this solution would only skip the first entry of possibly multiple chunk entries.
Let's say if entry with chunkId 0 got lost and there would be subsequent entries chunkId 1, chunkId 2 and chunkId 3. The entries with chunkId 2 and chunkId 3 would get delivered to the client, causing a similar issue.

[fix][broker]Fixed an issue where the entire subscription would be bl…

27ccbce

…ocked when a chunk message with an ID of zero did not exist.

github-actions bot added doc-label-missing doc-not-needed Your PR changes do not impact docs and removed doc-label-missing labels Dec 30, 2025

lhotari requested a review from Copilot January 2, 2026 09:55

Copilot started reviewing on behalf of lhotari January 2, 2026 09:56 View session

Copilot AI reviewed Jan 2, 2026

View reviewed changes

zjxxzjwang added 2 commits January 20, 2026 17:30

Merge branch 'refs/heads/master' into master-Chunk-message-repair2.0

d0eac18

# Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SharedConsumerAssignor.java

[improve][broker]Add an indicator for the count of expired messages w…

9d2d82f

…ithin a cycle

lhotari added the ready-to-test label Jan 20, 2026

lhotari added this to the 4.2.0 milestone Jan 20, 2026

lhotari added release/4.0.9 release/4.1.3 release/3.0.16 labels Jan 20, 2026

zjxxzjwang added 2 commits January 20, 2026 17:37

[improve][broker]Add an indicator for the count of expired messages w…

df84d98

…ithin a cycle

[improve][broker]Add an indicator for the count of expired messages w…

1cc432e

…ithin a cycle

[fix][broker] Fix chunked message loss when no consumers are available

5a17585

lhotari requested a review from Copilot January 27, 2026 11:04

Copilot started reviewing on behalf of lhotari January 27, 2026 11:05 View session

Copilot AI reviewed Jan 27, 2026

View reviewed changes

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SharedConsumerAssignor.java Outdated Show resolved Hide resolved

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SharedConsumerAssignorTest.java Outdated Show resolved Hide resolved

zjxxzjwang added 2 commits January 27, 2026 21:05

[fix][broker]Fixed an issue where the entire subscription would be bl…

ee76a33

…ocked when a chunk message with an ID of zero did not exist.

[fix][broker]Fixed an issue where the entire subscription would be bl…

6e2ab6e

…ocked when a chunk message with an ID of zero did not exist.

lhotari requested changes Feb 4, 2026

View reviewed changes

lhotari added release/3.0.17 and removed release/3.0.16 release/4.1.3 labels Feb 9, 2026

lhotari added release/4.1.4 release/4.0.10 and removed release/4.0.9 labels Feb 9, 2026

	private boolean skipChunk(EntryAndMetadata entryAndMetadata, MessageMetadata metadata) {
	private boolean shouldSkipChunk(EntryAndMetadata entryAndMetadata, MessageMetadata metadata) {

[fix][broker]Fixed an issue where the entire subscription would be blocked when a chunk message with an ID of zero did not exist. #25120

Are you sure you want to change the base?

[fix][broker]Fixed an issue where the entire subscription would be blocked when a chunk message with an ID of zero did not exist. #25120

Conversation

zjxxzjwang commented Dec 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Modifications

Documentation

Uh oh!

github-actions bot commented Dec 30, 2025

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

lhotari commented Jan 2, 2026

Uh oh!

zjxxzjwang commented Jan 20, 2026

Uh oh!

zjxxzjwang commented Jan 20, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

lhotari left a comment

Choose a reason for hiding this comment

Uh oh!

lhotari Feb 4, 2026

Choose a reason for hiding this comment

Uh oh!

lhotari Feb 4, 2026

Choose a reason for hiding this comment

Uh oh!

lhotari Feb 4, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

zjxxzjwang commented Dec 30, 2025 •

edited

Loading