Per-Message Nack Support With Redelivery Options

[yhl25]

Summary

Numaflow always acks messages today, even when users want a retry.
However, JetStream, Pulsar, and SQS already support nack, delayed redelivery, and max redeliveries.
We need to add a simple, consistent API so mappers and sinks can explicitly nack a message, similar to how they can drop or ok today.

Use Cases

• Users cannot nack messages in UDFs or sinks.
• Returning a response (including drop) still results in an ack.
• No way to request redelivery after a delay or limit max retries.

Proposed solution

Proposed API

1. NackOptions

NackOptions {
    Delay: time.Duration (optional)
    MaxDeliveries: int (optional)
    Reason: string (optional)
}

2. Mapper Nack

mapper.MessageToNack(opts)

3. Sink Nack

sinksdk.ResponseNack(id, opts)

If the backend does not support certain options, Numaflow should ignore them gracefully and log a warning.

Example Usage

Mapper Example

if err != nil {
    return mapper.MessagesBuilder().Append(
        mapper.MessageToNack(&mapper.NackOptions{
            Delay:         5 * time.Second,
            MaxDeliveries: 3,
            Reason:        "invalid integer, retry later",
        }),
    )
}

Sink Example

if writeError {
    return sinksdk.ResponseNack(id, &sinksdk.NackOptions{
        Delay: 10 * time.Second,
        MaxDeliveries: 5,
        Reason: "temporary sink failure",
    })
}

These match the existing SDK style, e.g.:

mapper.MessageToDrop()
sinksdk.ResponseOK(id)

---

Message from the maintainers:

If you wish to see this enhancement implemented please add a 👍 reaction to this issue! We often sort issues this way to know what to prioritize.