[Bug] Shared subscription stalls forever on a chunked message whose first chunk is already acked
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
### Search before reporting
- [x] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.
### Read release policy
- [x] I understand that [unsupported versions](https://pulsar.apache.org/contribute/release-policy/#supported-versions) don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.
### User environment
Broker 4.0.13, Go client consumer on a Shared subscription. The code looks the same on master.
### Issue Description
Producer has chunking enabled (messages can be tens of MB, so most of them get chunked),
consumed by a Shared subscription.
1. Publish a chunked message, chunks 0..2 land on entries 100, 101, 102.
2. Consumer gets chunks 0 and 1 and acks them, mark-delete moves to 101.
3. Consumer restarts. A topic unload or broker restart has the same effect.
4. Entry 102 (chunk 2) is next in line.
From that point the subscription never delivers anything again. Backlog grows, no error, no
warning. `pulsar-admin topics skip` on that one entry and everything resumes instantly.
**What did you expect to see**
The leftover chunk either delivered (clients already know how to discard an incomplete chunked
message) or skipped, and the rest of the topic dispatched normally.
**What did you see instead**
Nothing in the broker logs at all, which is why this took us so long to find.
`SharedConsumerAssignor.getConsumerForUuid()` looks the uuid up in `uuidToConsumer`. That map is
in-memory per dispatcher and only ever written when chunk 0 is assigned, so after the restart it
is empty and chunk 2 hits:
```java
if (metadata.getChunkId() != 0) {
// Not the first chunk, skip it
return null;
}
```
`null` makes `assign()` hand the entry to `unassignedMessageProcessor`, which the dispatcher wires
to `addEntryToReplay`, so it goes into `redeliveryMessages`. Next round the exact same thing
happens, because chunk 0 is acked and will never be read again to recreate the mapping. The entry stays in the replay queue.
### Error messages
```text
```
### Reproducing the issue
I created a a unittest and patch.
### Additional information
Please backport to 4.0 LTS
### Are you willing to submit a PR?
- [x] I'm willing to submit a PR!
Contributor guide
Research direction
Start with SharedConsumerAssignor.getConsumerForUuid() and trace how assign() sends a null consumer to unassignedMessageProcessor and addEntryToReplay. Run the reported unittest and inspect the submitter's patch to reproduce the restart case. Done means an orphaned later chunk is delivered or skipped after chunk 0 was acked, allowing subsequent messages to dispatch normally.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go, java
- Domain
- distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 58/100