apache / apache/pulsar-client-go
[PIP] Add `Consumer.RedeliverUnacknowledgedMessages()`
- Dominant language
- Go
- Stars
- 745
- Forks
- 389
- Avg merge
- 3d 20h
- Merged PRs (30d)
- 3
Description
## Goal
Let users ask the broker to send all unacknowledged messages again.
## Why
The Java client has this method: `void redeliverUnacknowledgedMessages();`. The Go client does not have it. A user already uses this command internally (for nack and ack-timeout), but a user always sends it with a specific list of message IDs. A user never sends the "redeliver everything" version, and users have no way to trigger it. Calling `Nack` on each message is not always possible. Sometimes the app no longer knows which messages it was holding - for example after a restart, an out-of-memory event that cleared its caches, or a panic that dropped an in-memory buffer.
Typical uses:
- After a restart or deploy: "send me everything that's still in flight, so I don't miss anything."
- After the app got into a bad state and lost track of its messages.
- In tests: force a redelivery without waiting for NackRedeliveryDelay.
## Proposed API
Add one method to the Consumer interface:
```go
type Consumer interface {
// ...
// RedeliverUnacknowledgedMessages asks the broker to send again all messages
// this consumer has received but not yet acknowledged. It happens in the
// background and does not block.
//
// Any messages already waiting in the local queue are dropped and requested
// from the broker again, so the app will receive them again through
// Receive() or Chan().
//
// For Shared and KeyShared subscriptions, the resent messages may go to
// other consumers on the same subscription.
RedeliverUnacknowledgedMessages()
}
```
There is no error return, because Java has none and the command is fire-and-forget.
## How it will work
This matches what the Java client does:
1. Drop the messages waiting in the local queue, and count how many there were.
2. Send the redeliver command with no message IDs. This tells the broker to resend everything that is unacknowledged.
3. Give the broker back the same number of flow permits as the messages we dropped, so it has room to resend them.
The function is added to all consumer types (single, partitioned, multi-topic, regex, and zero-queue), the same way `Pause()` and `Resume()` were added recently (https://github.com/apache/pulsar-client-go/pull/1507).
## Edge cases
- Nothing in flight: does nothing, no error, no panic.
- Closed consumer: ignored, with a log line.
- Paused consumer: the command is still sent (pausing only stops new message permits, not control commands). No messages are delivered until `Resume()` is called.
## What this does not include
- No per-message version like Java's `redeliverUnacknowledgedMessages(Set)`. Go users already have `Nack` and `NackID` for that.
- No consumer-epoch support. The Go client does not use consumer epoch anywhere today, so we leave it out to stay consistent.
There is a small chance of seeing a few duplicate messages - the ones already passed to the delivery channel when the call is made. This is fine, since Pulsar already delivers at least once and the whole point of this call is "send it all again."
### References
- Java: `ConsumerImpl.redeliverUnacknowledgedMessages()` and `MultiTopicsConsumerImpl.redeliverUnacknowledgedMessages()`.
Contributor guide
Research direction
Start at the Consumer interface and the existing Pause() and Resume() implementations, then trace how Nack and NackID send broker commands and how local queues and flow permits are handled. Check the single, partitioned, multi-topic, regex, and zero-queue consumer types. Done means each exposes the method with the described redelivery, queue-drop, permit, pause, and closed-consumer behavior.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go
- Domain
- backend-api-design, distributed-systems
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 55/100