Thanks PenghuiLi, I will try to base this PIP to provide the corresponding code to implement.
LGTM +1(non-binding) -- Thanks Xiaolong Ran PengHui Li <peng...@apache.org> 于2021年10月15日周五 下午3:56写道： > https://github.com/apache/pulsar/issues/12379 > > --- Pasted here for quoting convenience --- > > ## Motivation > > Apache Pulsar supports the at-least-once message delivery semantic which > can tolerate the consumer failure such as the consumer write the data to > the database but the database might offline for a while, call an external > HTTP server but the HTTP server is not available or maybe the parts of the > consumer fault(can’t connect to the database or HTTP server). > > In general, the consumer is not able to process the message successfully, > what we can do for the above case is we can redeliver the message after > processing the message failure so that the message can be redelivered to > other consumers(for the Shared subscription). This is a frequently used > example: > > ``` > Message msg = consumer.receive(); > > try { > process(msg); > consumer.acknowledge(msg); > } catch (Exception e) { > consumer.negativeAcknowledge(msg); > } > ``` > > But you might don’t want to redeliver the message immediately, give > non-working services(HTTP server, Database) some time to recover to avoid > the extra overhead caused by too frequent retries. Currently, we can > specify a delay for the message redelivery by the negative acknowledgment. > > ``` > client.newConsumer() > .... > .negativeAckRedeliveryDelay(1, TimeUnit.SECONDS) > .subscribe(); > ``` > > But this is not flexible enough, so the proposal is to introduce a > redelivery backoff mechanism which we can achieve redelivery with different > delays according to the number of times the message is retried such as 1s, > 2s, 4s, 8s, 16s in the next 5 times message redelivery. > > -> reconsumeLater > > ## Approach > > The approach is to introduce a `NegativeAckRedeliveryBackoff` at the > client-side, users can specify a `NegativeAckRedeliveryBackoff` for a > consumer. And the client will provide an implementation > `NegativeAckRedeliveryExponentialBackoff`. > > The NegativeAckBackoff cannot be used with redelivery delay together, and > the default redelivery delay will not change. > > Users are also able to implement a specific `NegativeAckRedeliveryBackoff`, > For some frequently used backoff implementations, we should also support it > in pulsar clients to provide users with an out-of-the-box experience. > > Notice: the consumer crashes will trigger the redelivery of the unacked > message, this case will not respect the `NegativeAckRedeliveryBackoff`, > which means the message might get redelivered earlier than the delay time > from the backoff. > > ## API changes > > The new `NegativeAckBackoff` interface > ``` > interface NegativeAckBackoff { > > long next(int redeliveryCount); > > } > ``` > > A new method for building the consumer > ``` > client.newConsumer() > .... > .negativeAckRedeliveryBackoff(...) > .subscribe(); > ``` > > Notice: the `NegativeAckRedeliveryBackoff` will not work with > `consumer.negativeAcknowledge(MessageId messageId)` because we are not able > to get the redelivery count from the message ID. > > ## Compatibility > > The proposal will not introduce any compatibility issues. > > ## Tests Plan > > Unit tests & integration tests >
