Support Consumers Set Custom Retry Delay

[liudezhi2098]

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.

Proposed changes

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]

Thanks for the great work @liudezhi2098

[devinbost]

Hi @liudezhi2098 , thanks for your work on this. I'm trying to understand how I'd advise users regarding how to implement this feature.
The docs provide this example:

https://pulsar.apache.org/docs/en/concepts-messaging/#retry-letter-topic

Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
                .topic(topic)
                .subscriptionName("my-subscription")
                .subscriptionType(SubscriptionType.Shared)
                .enableRetry(true)
                .receiverQueueSize(100)
                .deadLetterPolicy(DeadLetterPolicy.builder()
                        .maxRedeliverCount(maxRedeliveryCount)
                        .retryLetterTopic("persistent://my-property/my-ns/my-subscription-custom-Retry")
                        .build())
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                .subscribe();

However, it's not clear where they'd set the delay between retries. Could you please clarify this?

[devinbost]

I found some examples in your unit tests. It would be good to also update the docs to reflect this.