From ca0e2e4650c1d92b091fab03033fd61d2755166d Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Fri, 17 Jun 2022 16:17:46 -0700 Subject: [PATCH] Avoid some promise object allocations when writing on channel --- .../service/PulsarCommandSenderImpl.java | 48 +++++++++---------- 1 file changed, 24 insertions(+), 24 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java index 4cdc0a76e8c96..57dcaff250df3 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PulsarCommandSenderImpl.java @@ -54,7 +54,7 @@ public void sendPartitionMetadataResponse(ServerError error, String errorMsg, lo BaseCommand command = Commands.newPartitionMetadataResponseCommand(error, errorMsg, requestId); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -62,7 +62,7 @@ public void sendPartitionMetadataResponse(int partitions, long requestId) { BaseCommand command = Commands.newPartitionMetadataResponseCommand(partitions, requestId); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -70,7 +70,7 @@ public void sendSuccessResponse(long requestId) { BaseCommand command = Commands.newSuccessCommand(requestId); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -78,7 +78,7 @@ public void sendErrorResponse(long requestId, ServerError error, String message) BaseCommand command = Commands.newErrorCommand(requestId, error, message); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -86,7 +86,7 @@ public void sendProducerSuccessResponse(long requestId, String producerName, Sch BaseCommand command = Commands.newProducerSuccessCommand(requestId, producerName, schemaVersion); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -97,7 +97,7 @@ public void sendProducerSuccessResponse(long requestId, String producerName, lon schemaVersion, topicEpoch, isProducerReady); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -107,7 +107,7 @@ public void sendSendReceiptResponse(long producerId, long sequenceId, long highe entryId); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -115,7 +115,7 @@ public void sendSendError(long producerId, long sequenceId, ServerError error, S BaseCommand command = Commands.newSendErrorCommand(producerId, sequenceId, error, errorMsg); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -125,7 +125,7 @@ public void sendGetTopicsOfNamespaceResponse(List topics, String topicsH filtered, changed, requestId); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -133,7 +133,7 @@ public void sendGetSchemaResponse(long requestId, SchemaInfo schema, SchemaVersi BaseCommand command = Commands.newGetSchemaResponseCommand(requestId, schema, version); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -141,7 +141,7 @@ public void sendGetSchemaErrorResponse(long requestId, ServerError error, String BaseCommand command = Commands.newGetSchemaResponseErrorCommand(requestId, error, errorMessage); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -149,7 +149,7 @@ public void sendGetOrCreateSchemaResponse(long requestId, SchemaVersion schemaVe BaseCommand command = Commands.newGetOrCreateSchemaResponseCommand(requestId, schemaVersion); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -158,7 +158,7 @@ public void sendGetOrCreateSchemaErrorResponse(long requestId, ServerError error Commands.newGetOrCreateSchemaResponseErrorCommand(requestId, error, errorMessage); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -166,7 +166,7 @@ public void sendConnectedResponse(int clientProtocolVersion, int maxMessageSize) BaseCommand command = Commands.newConnectedCommand(clientProtocolVersion, maxMessageSize); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -177,7 +177,7 @@ public void sendLookupResponse(String brokerServiceUrl, String brokerServiceUrlT authoritative, response, requestId, proxyThroughServiceUrl); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -185,7 +185,7 @@ public void sendLookupResponse(ServerError error, String errorMsg, long requestI BaseCommand command = Commands.newLookupErrorResponseCommand(error, errorMsg, requestId); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -201,12 +201,12 @@ public void sendActiveConsumerChange(long consumerId, boolean isActive) { @Override public void sendSuccess(long requestId) { - cnx.ctx().writeAndFlush(Commands.newSuccess(requestId)); + cnx.ctx().writeAndFlush(Commands.newSuccess(requestId), cnx.ctx().voidPromise()); } @Override public void sendError(long requestId, ServerError error, String message) { - cnx.ctx().writeAndFlush(Commands.newError(requestId, error, message)); + cnx.ctx().writeAndFlush(Commands.newError(requestId, error, message), cnx.ctx().voidPromise()); } @Override @@ -214,7 +214,7 @@ public void sendReachedEndOfTopic(long consumerId) { // Only send notification if the client understand the command if (cnx.getRemoteEndpointProtocolVersion() >= ProtocolVersion.v9.getValue()) { log.info("[{}] Notifying consumer that end of topic has been reached", this); - cnx.ctx().writeAndFlush(Commands.newReachedEndOfTopic(consumerId)); + cnx.ctx().writeAndFlush(Commands.newReachedEndOfTopic(consumerId), cnx.ctx().voidPromise()); } } @@ -295,7 +295,7 @@ public void sendTcClientConnectResponse(long requestId, ServerError error, Strin BaseCommand command = Commands.newTcClientConnectResponse(requestId, error, message); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -309,7 +309,7 @@ public void sendNewTxnResponse(long requestId, TxnID txnID, long tcID) { txnID.getMostSigBits()); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); if (this.interceptor != null) { this.interceptor.txnOpened(tcID, txnID.toString()); } @@ -320,7 +320,7 @@ public void sendNewTxnErrorResponse(long requestId, long txnID, ServerError erro BaseCommand command = Commands.newTxnResponse(requestId, txnID, error, message); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } @Override @@ -329,7 +329,7 @@ public void sendEndTxnResponse(long requestId, TxnID txnID, int txnAction) { txnID.getMostSigBits()); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); if (this.interceptor != null) { this.interceptor.txnEnded(txnID.toString(), txnAction); } @@ -340,7 +340,7 @@ public void sendEndTxnErrorResponse(long requestId, TxnID txnID, ServerError err BaseCommand command = Commands.newEndTxnResponse(requestId, txnID.getMostSigBits(), error, message); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); if (this.interceptor != null) { this.interceptor.txnEnded(txnID.toString(), TxnAction.ABORT_VALUE); }