PavelZeger opened a new issue, #1510:
URL: https://github.com/apache/pulsar-client-go/issues/1510
## 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<MessageId>)`. 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()`.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
