Pulsar: Dead-Letter Queues

[sschepens]

Pulsar should support dead-letter queues to prevent consumers being stuck when they cannot acknowledge a message.
Reference #180

[agarman]

I am writing rejected messages to "TopicName{{RejectedMessage}}" then acknowledging the message. It'd be nice to have the broker handle this atomically.

An API like:

reject(msgId); // sends message into dead letter box for subscription; acknowledged in process
reroute(msgId, specificTopic); // same but into a specific topic

[merlimat]

The primary concern I have with dead-letter queue is how to avoid that some misconfiguration can lead to send everything into that topic.

In some cases the application can have a pretty good idea that a message is bad/corrupted and thus triggering the rerouting of the message, but if for example a new field is introduced and the consumer is not updated, it might start sending everything to DLQ.

About doing the work in client vs server. I don't think there is a big functional difference. The 2 operations cannot be atomically executed anyway, we have to first publish and then skip the bad message.

My preference would go to doing that in client lib, to avoid scalability concern in broker.

Adding on what proposed by @agarman, what about configuring the DLQ on consumer config?

consumerConf.setDeadLetterQueue("persistent://.....");
consumerConf.sendToDeadLetterQueueAfterRetries(10);

Then in consumer code:

while (true) {
    Message msg = consumer.receive();
    try {
        // processMessage()
    } catch (Exception e) {
       consumer.reject(msg);
       continue;
    }

   consumer.acknowledge(msg);
}

The difficult part would be how to support transient vs persistent errors and rejections.

[sijie]

PIP-22 is the proposal for implementing dead letter queue.

[sijie]

@codelipenghui @jiazhai : you guys can probably use this issue as the main issue for dead letter topic implementation.

[sijie]

2400 is implementing dead letter topic.

[ivankelly]

Closing as a dupe of #2400

[fracasula]

Hey guys, are there any plans to add DLQ support in the Golang CGo based library? I see that it's supported in the pure Golang one which we are not using though due to your disclaimer saying that feature parity and stability has not been reached.