lhotari commented on code in PR #23157:
URL: https://github.com/apache/pulsar/pull/23157#discussion_r1715442825
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -659,29 +660,8 @@ protected CompletableFuture<Void>
doReconsumeLater(Message<?> message, AckType a
}
return FutureUtil.failedFuture(exception);
}
- if (delayTime < 0) {
- delayTime = 0;
- }
- if (retryLetterProducer == null) {
- createProducerLock.writeLock().lock();
- try {
- if (retryLetterProducer == null) {
- retryLetterProducer =
client.newProducer(Schema.AUTO_PRODUCE_BYTES(schema))
- .topic(this.deadLetterPolicy.getRetryLetterTopic())
- .enableBatching(false)
- .enableChunking(true)
- .blockIfQueueFull(false)
- .create();
-
stats.setRetryLetterProducerStats(retryLetterProducer.getStats());
- }
- } catch (Exception e) {
- log.error("Create retry letter producer exception with topic:
{}",
- deadLetterPolicy.getRetryLetterTopic(), e);
- return FutureUtil.failedFuture(e);
Review Comment:
Previously the error handling was handled this way. I think that it would be
useful to preserve error handling for the case where creating the
retryLetterProducer fails. One possible location is to chain it after
`retryLetterProducer.thenAcceptAsync(...)` as suggested in a comment below.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
