Support Consumers Set Custom Retry Delay

[liudezhi2098]

Master Issue: #6448

Motivation

For many online business systems, various exceptions usually occur in business logic processing, so the message needs to be re-consumed, but users hope that this delay time can be controlled flexibly. The current user's processing method is usually to send this message to a special retry topic, because production can specify any delay, so consumers subscribe the business topic and retry topic at the same time. I think this logic can be supported by pulsar itself, making it easier for users to use, and it looks like this is a very common requirement.

Modifications

This change can be supported on the client side, need to add a set of interfaces to org.apache.pulsar.client.api.Consumer

void reconsumeLater(Message<?> message, long delayTime, TimeUnit unit) throws PulsarClientException;
CompletableFuture<Void> reconsumeLaterAsync(Message<?> message, long delayTime, TimeUnit unit);
CompletableFuture<Void> reconsumeLaterAsync(Messages<?> messages, int delayLevel);

DeadLetterPolicy add retry topic

public class DeadLetterPolicy {

    /**
     * Maximum number of times that a message will be redelivered before being sent to the dead letter queue.
     */
    private int maxRedeliverCount;

    /**
     * Name of the retry topic where the failing messages will be sent.
     */
    private String retryLetterTopic;

    /**
     * Name of the dead topic where the failing messages will be sent.
     */
    private String deadLetterTopic;

}

org.apache.pulsar.client.impl.ConsumerImpl add a retry producer

  private volatile Producer<T> deadLetterProducer;

  private volatile Producer<T> retryLetterProducer;

Can specify whether to enable retry when creating a consumer，default unenable

    @Override
    public ConsumerBuilder<T> enableRetry(boolean retryEnable) {
        conf.setRetryEnable(retryEnable);
        return this;
    }

[jiazhai]

@liudezhi2098 Thanks for the great work. Would you please help add some tests for it?

[jiazhai]

/pulsarbot run-failure-checks

[jiazhai]

/pulsarbot run-failure-checks

[codelipenghui]

/pulsarbot run-failure-checks

[Anonymitaet]

Doc has been added #7151

[Lanayx]

Does anyone know what is the point of this line? The logic of sending message to dead letter queue is described within more than twenty lines below as well as message acknowledgment, why is it done twice?

[Lanayx]

@liudezhi2098 Hi, could you please comment on my question above?