You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/04/21 21:10:45 UTC
(pulsar) 01/02: [fix][broker] Prevent timed-out producer creation from racing with retry (#25460)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c451abf3d690bd4f226ac1dd740fd665ae2178b3
Author: sinan liu <li...@gmail.com>
AuthorDate: Sat Apr 4 06:11:47 2026 +0800
[fix][broker] Prevent timed-out producer creation from racing with retry (#25460)
(cherry picked from commit 56442c27907f3e80e4876404ff31063811a92fba)
---
.../src/main/java/org/apache/pulsar/broker/service/ServerCnx.java | 7 +++++++
1 file changed, 7 insertions(+)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index 9a835370f7d..de1e16a0d2f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -1875,6 +1875,13 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
ProducerAccessMode producerAccessMode,
Optional<Long> topicEpoch, boolean supportsPartialProducer,
CompletableFuture<Producer> producerFuture){
+ if (producerFuture.isCompletedExceptionally()) {
+ log.info("[{}] Skipped producer creation after timeout on client side. producerId={}, producerName={}",
+ remoteAddress, producerId, producerName);
+ producers.remove(producerId, producerFuture);
+ return;
+ }
+
CompletableFuture<Void> producerQueuedFuture = new CompletableFuture<>();
Producer producer = new Producer(topic, ServerCnx.this, producerId, producerName,
getPrincipal(), isEncrypted, metadata, schemaVersion, epoch,