[broker] fix `producerFuture.completeExceptionally` not called before `commandSender.sendErrorResponse` in `ServerCnx.handleProducer`
### Motivation
`producerFuture.completeExceptionally` is not called before `sendErrorResponse` when encryption check or schema check failed in `ServerCnx.handleProduce`.
### Modifications
Add `producerFuture.completeExceptionally` before `sendErrorResponse`
@@ -1217,7 +1217,9 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
if ((topic.isEncryptionRequired() || encryptionRequireOnProducer) && !isEncrypted) {
String msg = String.format("Encryption is required in %s", topicName);
log.warn("[{}] {}", remoteAddress, msg);
+ if (producerFuture.completeExceptionally(new ServerMetadataException(msg))) {
commandSender.sendErrorResponse(requestId, ServerError.MetadataError, msg);
+ }
producers.remove(producerId, producerFuture);
return;
}
@@ -1227,6 +1229,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
CompletableFuture<SchemaVersion> schemaVersionFuture = tryAddSchema(topic, schema);
schemaVersionFuture.exceptionally(exception -> {
+ if (producerFuture.completeExceptionally(exception)) {
String message = exception.getMessage();
if (exception.getCause() != null) {
message += (" caused by " + exception.getCause());
@@ -1234,6 +1237,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
commandSender.sendErrorResponse(requestId,
BrokerServiceException.getClientErrorCode(exception),
message);
+ }
producers.remove(producerId, producerFuture);
return null;
});
