From 7ad288949bbd753859d6c051bb82747811207451 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Wed, 18 Mar 2026 12:56:14 -0700 Subject: [PATCH] [fix][broker] Fix race condition in ServerCnx producer/consumer async callbacks The ServerCnx producers and consumers ConcurrentLongHashMap maps are designed with concurrencyLevel=1, assuming all accesses happen on the same Netty IO thread (ctx.executor()). However, most async callbacks in the producer and consumer lifecycle used synchronous variants (thenAccept, thenCompose, etc.) instead of async variants with ctx.executor(). The issue is how CompletableFuture chaining works without an explicit executor: - If the upstream future is NOT yet completed when thenCompose(fn) is called, fn is queued as a dependent and runs when the completing thread calls complete(). Multiple chained stages execute in registration order since the completing thread walks the chain sequentially. - If the upstream future is ALREADY completed when thenCompose(fn) is called, fn runs IMMEDIATELY on the calling thread, right there in the thenCompose call. It skips the queue entirely. This distinction causes the create/close/create producer race: Both createProducer1 and createProducer2 call getOrCreateTopic() and chain thenCompose(topic -> addProducer(...)). When the topic future is not yet completed, both callbacks are queued as dependents and execute in order. But if the topic future is already completed when createProducer2 chains on it, producer2's addProducer() runs immediately inline - potentially before producer1's cleanup from the close command has finished. Producer1 wins the race to register, producer2's addProducer() fails with "already connected", and the client never gets a success response. The fix: using thenComposeAsync(fn, ctx.executor()) forces fn to always be submitted to the executor's task queue, regardless of whether the future is already completed. This guarantees FIFO ordering - all stages go through the same queue, so the close handler runs before producer2's creation, and producer1's cleanup completes before producer2 tries to register. Changes: - handleProducer chain: thenApplyAsync, thenComposeAsync, thenRunAsync, exceptionallyAsync with ctx.executor() - buildProducerAndAddTopic: thenAcceptAsync, thenRunAsync - handleSubscribe chain: thenApplyAsync, thenAcceptAsync, exceptionallyAsync - handleCloseProducer: thenAcceptAsync - safelyRemoveProducer/Consumer: whenCompleteAsync - ServerCnxTest: add channel.runPendingTasks() in getResponse() polling loop so EmbeddedChannel executor tasks are processed --- .../pulsar/broker/service/ServerCnx.java | 62 +++++++++---------- .../pulsar/broker/service/ServerCnxTest.java | 7 ++- 2 files changed, 37 insertions(+), 32 deletions(-) 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 4b63dd05c2d31..f84f1c080ede3 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 @@ -1358,7 +1358,7 @@ protected void handleSubscribe(final CommandSubscribe subscribe) { CompletableFuture consumerFuture = new CompletableFuture<>(); CompletableFuture existingConsumerFuture = consumers.putIfAbsent(consumerId, consumerFuture); - isAuthorizedFuture.thenApply(isAuthorized -> { + isAuthorizedFuture.thenApplyAsync(isAuthorized -> { if (isAuthorized) { if (log.isDebugEnabled()) { log.debug("[{}] Client is authorized to subscribe with role {}", @@ -1490,7 +1490,7 @@ protected void handleSubscribe(final CommandSubscribe subscribe) { } }); }) - .thenAccept(consumer -> { + .thenAcceptAsync(consumer -> { if (consumer.checkAndApplyTopicMigration()) { log.info("[{}] Disconnecting consumer {} on migrated subscription on topic {} / {}", remoteAddress, consumerId, subscriptionName, topicName); @@ -1524,8 +1524,8 @@ protected void handleSubscribe(final CommandSubscribe subscribe) { consumers.remove(consumerId, consumerFuture); } - }) - .exceptionally(exception -> { + }, ctx.executor()) + .exceptionallyAsync(exception -> { if (exception.getCause() instanceof ConsumerBusyException) { if (log.isDebugEnabled()) { log.debug( @@ -1573,7 +1573,7 @@ protected void handleSubscribe(final CommandSubscribe subscribe) { return null; - }); + }, ctx.executor()); } else { String msg = "Client is not authorized to subscribe"; log.warn("[{}] {} with role {}", remoteAddress, msg, getPrincipal()); @@ -1581,12 +1581,12 @@ protected void handleSubscribe(final CommandSubscribe subscribe) { writeAndFlush(Commands.newError(requestId, ServerError.AuthorizationError, msg)); } return null; - }).exceptionally(ex -> { + }, ctx.executor()).exceptionallyAsync(ex -> { logAuthException(remoteAddress, "subscribe", getPrincipal(), Optional.of(topicName), ex); consumers.remove(consumerId, consumerFuture); commandSender.sendErrorResponse(requestId, ServerError.AuthorizationError, ex.getMessage()); return null; - }); + }, ctx.executor()); } private SchemaData getSchema(Schema protocolSchema) { @@ -1642,7 +1642,7 @@ protected void handleProducer(final CommandProducer cmdProducer) { (canProduce, canSubscribe) -> canProduce && canSubscribe); } - isAuthorizedFuture.thenApply(isAuthorized -> { + isAuthorizedFuture.thenApplyAsync(isAuthorized -> { if (!isAuthorized) { String msg = "Client is not authorized to Produce"; log.warn("[{}] {} with role {}", remoteAddress, msg, getPrincipal()); @@ -1689,7 +1689,7 @@ protected void handleProducer(final CommandProducer cmdProducer) { topicName, producerId, producerName, schema == null ? "absent" : "present"); } - service.getOrCreateTopic(topicName.toString()).thenCompose((Topic topic) -> { + service.getOrCreateTopic(topicName.toString()).thenComposeAsync((Topic topic) -> { // Check max producer limitation to avoid unnecessary ops wasting resources. For example: the new // producer reached max producer limitation, but pulsar did schema check first, it would waste CPU if (((AbstractTopic) topic).isProducersExceeded(producerName)) { @@ -1705,7 +1705,7 @@ protected void handleProducer(final CommandProducer cmdProducer) { topic.checkBacklogQuotaExceeded(producerName, BacklogQuotaType.destination_storage), topic.checkBacklogQuotaExceeded(producerName, BacklogQuotaType.message_age)); - backlogQuotaCheckFuture.thenRun(() -> { + backlogQuotaCheckFuture.thenRunAsync(() -> { // Check whether the producer will publish encrypted messages or not if ((topic.isEncryptionRequired() || encryptionRequireOnProducer) && !isEncrypted @@ -1723,7 +1723,7 @@ protected void handleProducer(final CommandProducer cmdProducer) { CompletableFuture schemaVersionFuture = tryAddSchema(topic, schema); - schemaVersionFuture.exceptionally(exception -> { + schemaVersionFuture.exceptionallyAsync(exception -> { if (producerFuture.completeExceptionally(exception)) { String message = exception.getMessage(); if (exception.getCause() != null) { @@ -1747,9 +1747,9 @@ protected void handleProducer(final CommandProducer cmdProducer) { } producers.remove(producerId, producerFuture); return null; - }); + }, ctx.executor()); - schemaVersionFuture.thenAccept(schemaVersion -> { + schemaVersionFuture.thenAcceptAsync(schemaVersion -> { CompletionStage createInitSubFuture; if (!Strings.isNullOrEmpty(initialSubscriptionName) && topic.isPersistent() @@ -1769,7 +1769,7 @@ protected void handleProducer(final CommandProducer cmdProducer) { createInitSubFuture = CompletableFuture.completedFuture(null); } - createInitSubFuture.whenComplete((sub, ex) -> { + createInitSubFuture.whenCompleteAsync((sub, ex) -> { if (ex != null) { final Throwable rc = FutureUtil.unwrapCompletionException(ex); if (rc instanceof BrokerServiceException.NotAllowedException) { @@ -1797,11 +1797,11 @@ protected void handleProducer(final CommandProducer cmdProducer) { buildProducerAndAddTopic(topic, producerId, producerName, requestId, isEncrypted, metadata, schemaVersion, epoch, userProvidedProducerName, topicName, producerAccessMode, topicEpoch, supportsPartialProducer, producerFuture); - }); - }); - }); + }, ctx.executor()); + }, ctx.executor()); + }, ctx.executor()); return backlogQuotaCheckFuture; - }).exceptionally(exception -> { + }, ctx.executor()).exceptionallyAsync(exception -> { Throwable cause = exception.getCause(); if (cause instanceof BrokerServiceException.TopicBacklogQuotaExceededException) { BrokerServiceException.TopicBacklogQuotaExceededException tbqe = @@ -1860,13 +1860,13 @@ protected void handleProducer(final CommandProducer cmdProducer) { } producers.remove(producerId, producerFuture); return null; - }); + }, ctx.executor()); return null; - }).exceptionally(ex -> { + }, ctx.executor()).exceptionallyAsync(ex -> { logAuthException(remoteAddress, "producer", getPrincipal(), Optional.of(topicName), ex); commandSender.sendErrorResponse(requestId, ServerError.AuthorizationError, ex.getMessage()); return null; - }); + }, ctx.executor()); } private void buildProducerAndAddTopic(Topic topic, long producerId, String producerName, long requestId, @@ -1880,7 +1880,7 @@ private void buildProducerAndAddTopic(Topic topic, long producerId, String produ getPrincipal(), isEncrypted, metadata, schemaVersion, epoch, userProvidedProducerName, producerAccessMode, topicEpoch, supportsPartialProducer); - topic.addProducer(producer, producerQueuedFuture).thenAccept(newTopicEpoch -> { + topic.addProducer(producer, producerQueuedFuture).thenAcceptAsync(newTopicEpoch -> { if (isActive()) { if (producerFuture.complete(producer)) { log.info("[{}] Created new producer: {}, role: {}", remoteAddress, producer, getPrincipal()); @@ -1913,7 +1913,7 @@ private void buildProducerAndAddTopic(Topic topic, long producerId, String produ } producers.remove(producerId, producerFuture); - }).exceptionallyAsync(ex -> { + }, ctx.executor()).exceptionallyAsync(ex -> { if (ex.getCause() instanceof BrokerServiceException.TopicMigratedException) { Optional clusterURL = getMigratedClusterUrl(service.getPulsar(), topic.getName()); if (clusterURL.isPresent()) { @@ -1956,7 +1956,7 @@ private void buildProducerAndAddTopic(Topic topic, long producerId, String produ return null; }, ctx.executor()); - producerQueuedFuture.thenRun(() -> { + producerQueuedFuture.thenRunAsync(() -> { // If the producer is queued waiting, we will get an immediate notification // that we need to pass to client if (isActive()) { @@ -1969,7 +1969,7 @@ private void buildProducerAndAddTopic(Topic topic, long producerId, String produ producerCreated(this, producer, metadata); } } - }); + }, ctx.executor()); } @Override protected void handleSend(CommandSend send, ByteBuf headersAndPayload) { @@ -2299,7 +2299,7 @@ protected void handleCloseProducer(CommandCloseProducer closeProducer) { log.info("[{}][{}] Closing producer on cnx {}. producerId={}", producer.getTopic(), producer.getProducerName(), remoteAddress, producerId); - producer.close(true).thenAccept(v -> { + producer.close(true).thenAcceptAsync(v -> { log.info("[{}][{}] Closed producer on cnx {}. producerId={}", producer.getTopic(), producer.getProducerName(), remoteAddress, producerId); @@ -2308,7 +2308,7 @@ protected void handleCloseProducer(CommandCloseProducer closeProducer) { if (brokerInterceptor != null) { brokerInterceptor.producerClosed(this, producer, producer.getMetadata()); } - }); + }, ctx.executor()); } @Override @@ -3439,11 +3439,11 @@ private void safelyRemoveProducer(Producer producer) { } CompletableFuture future = producers.get(producerId); if (future != null) { - future.whenComplete((producer2, exception) -> { + future.whenCompleteAsync((producer2, exception) -> { if (exception != null || producer2 == producer) { producers.remove(producerId, future); } - }); + }, ctx.executor()); } } @@ -3454,11 +3454,11 @@ private void safelyRemoveConsumer(Consumer consumer) { } CompletableFuture future = consumers.get(consumerId); if (future != null) { - future.whenComplete((consumer2, exception) -> { + future.whenCompleteAsync((consumer2, exception) -> { if (exception != null || consumer2 == consumer) { consumers.remove(consumerId, future); } - }); + }, ctx.executor()); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java index f0907031882e7..2151eacd5d63d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java @@ -2381,7 +2381,10 @@ public void testSubscribeTimeout() throws Exception { assertEquals(((CommandError) response).getRequestId(), 5); // We should receive response for 1st producer, since it was not cancelled by the close - Awaitility.await().untilAsserted(() -> assertFalse(channel.outboundMessages().isEmpty())); + Awaitility.await().untilAsserted(() -> { + channel.runPendingTasks(); + assertFalse(channel.outboundMessages().isEmpty()); + }); assertTrue(channel.isActive()); response = getResponse(); @@ -2889,6 +2892,8 @@ protected Object getResponse(EmbeddedChannel channel, ClientChannelHelper client final long sleepTimeMs = 10; final long iterations = TimeUnit.SECONDS.toMillis(10) / sleepTimeMs; for (int i = 0; i < iterations; i++) { + // Execute tasks submitted to ctx.executor() via thenAcceptAsync/thenRunAsync etc. + channel.runPendingTasks(); if (!channel.outboundMessages().isEmpty()) { Object outObject = channel.outboundMessages().remove(); Object cmd = clientChannelHelper.getCommand(outObject);