Skip to content

[fix][broker] Fix race condition in ServerCnx producer/consumer async callbacks #25352

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

Merged
lhotari merged 2 commits into from
Mar 19, 2026
Merged

[fix][broker] Fix race condition in ServerCnx producer/consumer async callbacks #25352

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7ad2889
[fix][broker] Fix race condition in ServerCnx producer/consumer async…
merlimat Mar 18, 2026
cc60065
Merge remote-tracking branch 'apache/master' into fix/ServerCnx-async…
merlimat Mar 19, 2026
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
62 changes: 31 additions & 31 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1358,7 +1358,7 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
CompletableFuture<Consumer> consumerFuture = new CompletableFuture<>();
CompletableFuture<Consumer> 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 {}",
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -1573,20 +1573,20 @@ 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());
consumers.remove(consumerId, consumerFuture);
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) {
Expand Down Expand Up @@ -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());
Expand Down Expand Up @@ -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)) {
Expand All @@ -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
Expand All @@ -1723,7 +1723,7 @@ protected void handleProducer(final CommandProducer cmdProducer) {

CompletableFuture<SchemaVersion> schemaVersionFuture = tryAddSchema(topic, schema);

schemaVersionFuture.exceptionally(exception -> {
schemaVersionFuture.exceptionallyAsync(exception -> {
if (producerFuture.completeExceptionally(exception)) {
String message = exception.getMessage();
if (exception.getCause() != null) {
Expand All @@ -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<Subscription> createInitSubFuture;
if (!Strings.isNullOrEmpty(initialSubscriptionName)
&& topic.isPersistent()
Expand All @@ -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) {
Expand Down Expand Up @@ -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 =
Expand Down Expand Up @@ -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,
Expand All @@ -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());
Expand Down Expand Up @@ -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> clusterURL = getMigratedClusterUrl(service.getPulsar(), topic.getName());
if (clusterURL.isPresent()) {
Expand Down Expand Up @@ -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()) {
Expand All @@ -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) {
Expand Down Expand Up @@ -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);
Expand All @@ -2308,7 +2308,7 @@ protected void handleCloseProducer(CommandCloseProducer closeProducer) {
if (brokerInterceptor != null) {
brokerInterceptor.producerClosed(this, producer, producer.getMetadata());
}
});
}, ctx.executor());
}

@Override
Expand Down Expand Up @@ -3439,11 +3439,11 @@ private void safelyRemoveProducer(Producer producer) {
}
CompletableFuture<Producer> 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());
}
}

Expand All @@ -3454,11 +3454,11 @@ private void safelyRemoveConsumer(Consumer consumer) {
}
CompletableFuture<Consumer> 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());
}
}

Expand Down
7 changes: 6 additions & 1 deletion pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down Expand Up @@ -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);
Expand Down
Loading