Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[fix][client] Create the retry producer async #23157

Merged
merged 3 commits into from
Aug 14, 2024
Merged

[fix][client] Create the retry producer async #23157

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8d64e04
[fix][client] Create the retry producer async
Aug 12, 2024
f634ded
Fix retry letter producer stats
lhotari Aug 13, 2024
4c5f92d
Propagate failure when Retry Letter producer creation fails
Aug 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
96 changes: 52 additions & 44 deletions pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -203,7 +203,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle

private volatile CompletableFuture<Producer<byte[]>> deadLetterProducer;

private volatile Producer<byte[]> retryLetterProducer;
private volatile CompletableFuture<Producer<byte[]>> retryLetterProducer;
private final ReadWriteLock createProducerLock = new ReentrantReadWriteLock();

protected volatile boolean paused;
Expand Down Expand Up @@ -643,6 +643,7 @@ protected CompletableFuture<Void> doReconsumeLater(Message<?> message, AckType a
Map<String, String> customProperties,
long delayTime,
TimeUnit unit) {

MessageId messageId = message.getMessageId();
if (messageId == null) {
return FutureUtil.failedFuture(new PulsarClientException
Expand All @@ -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);
lhotari marked this conversation as resolved.
Show resolved Hide resolved
} finally {
createProducerLock.writeLock().unlock();
}
}

initRetryLetterProducerIfNeeded();
CompletableFuture<Void> result = new CompletableFuture<>();
if (retryLetterProducer != null) {
try {
Expand All @@ -701,7 +681,7 @@ protected CompletableFuture<Void> doReconsumeLater(Message<?> message, AckType a
}
propertiesMap.put(RetryMessageUtil.SYSTEM_PROPERTY_RECONSUMETIMES, String.valueOf(reconsumeTimes));
propertiesMap.put(RetryMessageUtil.SYSTEM_PROPERTY_DELAY_TIME,
String.valueOf(unit.toMillis(delayTime)));
String.valueOf(unit.toMillis(delayTime < 0 ? 0 : delayTime)));

MessageId finalMessageId = messageId;
if (reconsumeTimes > this.deadLetterPolicy.getMaxRedeliverCount()
Expand Down Expand Up @@ -732,23 +712,29 @@ protected CompletableFuture<Void> doReconsumeLater(Message<?> message, AckType a
});
} else {
assert retryMessage != null;
TypedMessageBuilder<byte[]> typedMessageBuilderNew = retryLetterProducer
.newMessage(Schema.AUTO_PRODUCE_BYTES(message.getReaderSchema().get()))
.value(retryMessage.getData())
.properties(propertiesMap);
if (delayTime > 0) {
typedMessageBuilderNew.deliverAfter(delayTime, unit);
}
if (message.hasKey()) {
typedMessageBuilderNew.key(message.getKey());
}
typedMessageBuilderNew.sendAsync()
.thenCompose(__ -> doAcknowledge(finalMessageId, ackType, Collections.emptyMap(), null))
.thenAccept(v -> result.complete(null))
.exceptionally(ex -> {
result.completeExceptionally(ex);
return null;
});
retryLetterProducer.thenAcceptAsync(rtlProducer -> {
TypedMessageBuilder<byte[]> typedMessageBuilderNew = rtlProducer
.newMessage(Schema.AUTO_PRODUCE_BYTES(message.getReaderSchema().get()))
.value(retryMessage.getData())
.properties(propertiesMap);
if (delayTime > 0) {
typedMessageBuilderNew.deliverAfter(delayTime, unit);
}
if (message.hasKey()) {
typedMessageBuilderNew.key(message.getKey());
}
typedMessageBuilderNew.sendAsync()
.thenCompose(__ -> doAcknowledge(finalMessageId, ackType, Collections.emptyMap(), null))
.thenAccept(v -> result.complete(null))
.exceptionally(ex -> {
result.completeExceptionally(ex);
return null;
});
}, internalPinnedExecutor).exceptionally(ex -> {
result.completeExceptionally(ex);
retryLetterProducer = null;
return null;
});
}
} catch (Exception e) {
result.completeExceptionally(e);
Expand All @@ -757,7 +743,7 @@ protected CompletableFuture<Void> doReconsumeLater(Message<?> message, AckType a
MessageId finalMessageId = messageId;
result.exceptionally(ex -> {
log.error("Send to retry letter topic exception with topic: {}, messageId: {}",
retryLetterProducer.getTopic(), finalMessageId, ex);
this.deadLetterPolicy.getRetryLetterTopic(), finalMessageId, ex);
Set<MessageId> messageIds = Collections.singleton(finalMessageId);
unAckedMessageTracker.remove(finalMessageId);
redeliverUnacknowledgedMessages(messageIds);
Expand Down Expand Up @@ -1136,7 +1122,7 @@ public synchronized CompletableFuture<Void> closeAsync() {
ArrayList<CompletableFuture<Void>> closeFutures = new ArrayList<>(4);
closeFutures.add(closeFuture);
if (retryLetterProducer != null) {
closeFutures.add(retryLetterProducer.closeAsync().whenComplete((ignore, ex) -> {
closeFutures.add(retryLetterProducer.thenCompose(p -> p.closeAsync()).whenComplete((ignore, ex) -> {
if (ex != null) {
log.warn("Exception ignored in closing retryLetterProducer of consumer", ex);
}
Expand Down Expand Up @@ -2267,6 +2253,28 @@ private void initDeadLetterProducerIfNeeded() {
}
}

private void initRetryLetterProducerIfNeeded() {
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)
.createAsync();
retryLetterProducer.thenAccept(rtlProducer -> {
stats.setRetryLetterProducerStats(rtlProducer.getStats());
});
}
} finally {
createProducerLock.writeLock().unlock();
}
}
}

@Override
public void seek(MessageId messageId) throws PulsarClientException {
try {
Expand Down
Loading