From f1c7444c39731f5f47cea4365a34c505ddd1cae1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=B2=20Boschi?= Date: Tue, 10 Jan 2023 10:58:43 +0100 Subject: [PATCH 1/4] =?UTF-8?q?[improve][broker,proxy]=C2=A0Use=20ChannelV?= =?UTF-8?q?oidPromise=20to=20avoid=20useless=20promise=20objects=20creatio?= =?UTF-8?q?n?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/PulsarCommandSenderImpl.java | 57 ++++----- .../pulsar/broker/service/ServerCnx.java | 116 +++++++++--------- .../pulsar/common/protocol/PulsarDecoder.java | 21 ++-- .../proxy/server/LookupProxyHandler.java | 45 ++++--- 4 files changed, 125 insertions(+), 114 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 e7cc25b2c3b89..0b489f42e5d5f 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 @@ -55,7 +55,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -63,7 +63,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -71,7 +71,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -79,7 +79,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -87,7 +87,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -98,7 +98,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -108,7 +108,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -116,7 +116,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -126,7 +126,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -134,7 +134,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -142,7 +142,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -150,7 +150,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -159,7 +159,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -168,7 +168,7 @@ public void sendConnectedResponse(int clientProtocolVersion, int maxMessageSize, clientProtocolVersion, maxMessageSize, supportsTopicWatchers); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -179,7 +179,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -187,7 +187,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -196,9 +196,7 @@ public void sendActiveConsumerChange(long consumerId, boolean isActive) { // if the client is older than `v12`, we don't need to send consumer group changes. return; } - cnx.ctx().writeAndFlush( - Commands.newActiveConsumerChange(consumerId, isActive), - cnx.ctx().voidPromise()); + writeAndFlushVoidPromise(Commands.newActiveConsumerChange(consumerId, isActive)); } @Override @@ -206,7 +204,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().voidPromise()); + writeAndFlushVoidPromise(Commands.newReachedEndOfTopic(consumerId)); } } @@ -215,8 +213,7 @@ public boolean sendTopicMigrated(ResourceType type, long resourceId, String brok // Only send notification if the client understand the command if (cnx.getRemoteEndpointProtocolVersion() >= ProtocolVersion.v20.getValue()) { log.info("[{}] Notifying {} that topic is migrated", type.name(), resourceId); - cnx.ctx().writeAndFlush(Commands.newTopicMigrated(type, resourceId, brokerUrl, brokerUrlTls), - cnx.ctx().voidPromise()); + writeAndFlushVoidPromise(Commands.newTopicMigrated(type, resourceId, brokerUrl, brokerUrlTls)); return true; } return false; @@ -310,7 +307,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -324,7 +321,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); if (this.interceptor != null) { this.interceptor.txnOpened(tcID, txnID.toString()); } @@ -335,7 +332,7 @@ public void sendNewTxnErrorResponse(long requestId, long tcID, ServerError error BaseCommand command = Commands.newTxnResponse(requestId, tcID, error, message); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); + writeAndFlushVoidPromise(outBuf); } @Override @@ -344,7 +341,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().voidPromise()); + writeAndFlushVoidPromise(outBuf); if (this.interceptor != null) { this.interceptor.txnEnded(txnID.toString(), txnAction); } @@ -356,7 +353,7 @@ public void sendEndTxnErrorResponse(long requestId, TxnID txnID, ServerError err txnID.getMostSigBits(), error, message); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); + writeAndFlushVoidPromise(outBuf); if (this.interceptor != null) { this.interceptor.txnEnded(txnID.toString(), TxnAction.ABORT_VALUE); } @@ -378,7 +375,11 @@ public void sendWatchTopicListUpdate(long watcherId, private void interceptAndWriteCommand(BaseCommand command) { safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - cnx.ctx().writeAndFlush(outBuf); + writeAndFlushVoidPromise(outBuf); + } + + private void writeAndFlushVoidPromise(ByteBuf outBuf) { + cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); } private void safeIntercept(BaseCommand command, ServerCnx cnx) { 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 a61468c6469ab..49b125e069247 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 @@ -480,7 +480,7 @@ protected void handleLookup(CommandLookupTopic lookup) { log.debug("[{}] Failed lookup topic {} due to pulsar service is not ready: {} state", remoteAddress, topicName, this.service.getPulsar().getState().toString()); } - ctx.writeAndFlush(newLookupErrorResponse(ServerError.ServiceNotReady, + writeAndFlushVoidPromise(newLookupErrorResponse(ServerError.ServiceNotReady, "Failed due to pulsar service is not ready", requestId)); return; } @@ -491,7 +491,7 @@ protected void handleLookup(CommandLookupTopic lookup) { final String msg = "Valid Proxy Client role should be provided for lookup "; log.warn("[{}] {} with role {} and proxyClientAuthRole {} on topic {}", remoteAddress, msg, authRole, originalPrincipal, topicName); - ctx.writeAndFlush(newLookupErrorResponse(ServerError.AuthorizationError, msg, requestId)); + writeAndFlushVoidPromise(newLookupErrorResponse(ServerError.AuthorizationError, msg, requestId)); lookupSemaphore.release(); return; } @@ -502,12 +502,12 @@ protected void handleLookup(CommandLookupTopic lookup) { getPrincipal(), getAuthenticationData(), requestId, advertisedListenerName).handle((lookupResponse, ex) -> { if (ex == null) { - ctx.writeAndFlush(lookupResponse); + writeAndFlushVoidPromise(lookupResponse); } else { // it should never happen log.warn("[{}] lookup failed with error {}, {}", remoteAddress, topicName, ex.getMessage(), ex); - ctx.writeAndFlush(newLookupErrorResponse(ServerError.ServiceNotReady, + writeAndFlushVoidPromise(newLookupErrorResponse(ServerError.ServiceNotReady, ex.getMessage(), requestId)); } lookupSemaphore.release(); @@ -516,14 +516,14 @@ protected void handleLookup(CommandLookupTopic lookup) { } else { final String msg = "Client is not authorized to Lookup"; log.warn("[{}] {} with role {} on topic {}", remoteAddress, msg, getPrincipal(), topicName); - ctx.writeAndFlush(newLookupErrorResponse(ServerError.AuthorizationError, msg, requestId)); + writeAndFlushVoidPromise(newLookupErrorResponse(ServerError.AuthorizationError, msg, requestId)); lookupSemaphore.release(); } return null; }).exceptionally(ex -> { logAuthException(remoteAddress, "lookup", getPrincipal(), Optional.of(topicName), ex); final String msg = "Exception occurred while trying to authorize lookup"; - ctx.writeAndFlush(newLookupErrorResponse(ServerError.AuthorizationError, msg, requestId)); + writeAndFlushVoidPromise(newLookupErrorResponse(ServerError.AuthorizationError, msg, requestId)); lookupSemaphore.release(); return null; }); @@ -531,11 +531,15 @@ protected void handleLookup(CommandLookupTopic lookup) { if (log.isDebugEnabled()) { log.debug("[{}] Failed lookup due to too many lookup-requests {}", remoteAddress, topicName); } - ctx.writeAndFlush(newLookupErrorResponse(ServerError.TooManyRequests, + writeAndFlushVoidPromise(newLookupErrorResponse(ServerError.TooManyRequests, "Failed due to too many pending lookup requests", requestId)); } } + private void writeAndFlushVoidPromise(ByteBuf cmd) { + ctx.writeAndFlush(cmd, ctx.voidPromise()); + } + @Override protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata partitionMetadata) { checkArgument(state == State.Connected); @@ -557,7 +561,7 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa partitionMetadata.getTopic(), remoteAddress, requestId, this.service.getPulsar().getState().toString()); } - ctx.writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.ServiceNotReady, + writeAndFlushVoidPromise(Commands.newPartitionMetadataResponse(ServerError.ServiceNotReady, "Failed due to pulsar service is not ready", requestId)); return; } @@ -607,15 +611,14 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa } else { final String msg = "Client is not authorized to Get Partition Metadata"; log.warn("[{}] {} with role {} on topic {}", remoteAddress, msg, getPrincipal(), topicName); - ctx.writeAndFlush( - Commands.newPartitionMetadataResponse(ServerError.AuthorizationError, msg, requestId)); + writeAndFlushVoidPromise(Commands.newPartitionMetadataResponse(ServerError.AuthorizationError, msg, requestId)); lookupSemaphore.release(); } return null; }).exceptionally(ex -> { logAuthException(remoteAddress, "partition-metadata", getPrincipal(), Optional.of(topicName), ex); final String msg = "Exception occurred while trying to authorize get Partition Metadata"; - ctx.writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.AuthorizationError, msg, + writeAndFlushVoidPromise(Commands.newPartitionMetadataResponse(ServerError.AuthorizationError, msg, requestId)); lookupSemaphore.release(); return null; @@ -657,7 +660,7 @@ protected void handleConsumerStats(CommandConsumerStats commandConsumerStats) { msg = createConsumerStatsResponse(consumer, requestId); } - ctx.writeAndFlush(msg); + writeAndFlushVoidPromise(msg); } ByteBuf createConsumerStatsResponse(Consumer consumer, long requestId) { @@ -687,7 +690,7 @@ ByteBuf createConsumerStatsResponse(Consumer consumer, long requestId) { // complete the connect and sent newConnected command private void completeConnect(int clientProtoVersion, String clientVersion, boolean supportsTopicWatchers) { - ctx.writeAndFlush(Commands.newConnected(clientProtoVersion, maxMessageSize, supportsTopicWatchers)); + writeAndFlushVoidPromise(Commands.newConnected(clientProtoVersion, maxMessageSize, supportsTopicWatchers)); state = State.Connected; service.getPulsarStats().recordConnectionCreateSuccess(); if (log.isDebugEnabled()) { @@ -763,7 +766,7 @@ private State doAuthentication(AuthData clientData, } // auth not complete, continue auth with client side. - ctx.writeAndFlush(Commands.newAuthChallenge(authMethod, brokerData, clientProtocolVersion)); + writeAndFlushVoidPromise(Commands.newAuthChallenge(authMethod, brokerData, clientProtocolVersion)); if (log.isDebugEnabled()) { log.debug("[{}] Authentication in progress client by method {}.", remoteAddress, authMethod); @@ -813,7 +816,7 @@ public void refreshAuthenticationCredentials() { try { AuthData brokerData = authState.refreshAuthentication(); - ctx.writeAndFlush(Commands.newAuthChallenge(authMethod, brokerData, + writeAndFlushVoidPromise(Commands.newAuthChallenge(authMethod, brokerData, getRemoteEndpointProtocolVersion())); if (log.isDebugEnabled()) { log.debug("[{}] Sent auth challenge to client to refresh credentials with method: {}.", @@ -849,8 +852,7 @@ protected void handleConnect(CommandConnect connect) { log.debug("Failed CONNECT from {} due to pulsar service is not ready: {} state", remoteAddress, this.service.getPulsar().getState().toString()); } - ctx.writeAndFlush( - Commands.newError(-1, ServerError.ServiceNotReady, "Failed due to pulsar service is not ready")); + writeAndFlushVoidPromise(Commands.newError(-1, ServerError.ServiceNotReady, "Failed due to pulsar service is not ready")); close(); return; } @@ -961,7 +963,7 @@ protected void handleConnect(CommandConnect connect) { service.getPulsarStats().recordConnectionCreateFail(); logAuthException(remoteAddress, "connect", getPrincipal(), Optional.empty(), e); String msg = "Unable to authenticate"; - ctx.writeAndFlush(Commands.newError(-1, ServerError.AuthenticationError, msg)); + writeAndFlushVoidPromise(Commands.newError(-1, ServerError.AuthenticationError, msg)); close(); } } @@ -985,13 +987,13 @@ protected void handleAuthResponse(CommandAuthResponse authResponse) { } catch (AuthenticationException e) { service.getPulsarStats().recordConnectionCreateFail(); log.warn("[{}] Authentication failed: {} ", remoteAddress, e.getMessage()); - ctx.writeAndFlush(Commands.newError(-1, ServerError.AuthenticationError, e.getMessage())); + writeAndFlushVoidPromise(Commands.newError(-1, ServerError.AuthenticationError, e.getMessage())); close(); } catch (Exception e) { service.getPulsarStats().recordConnectionCreateFail(); String msg = "Unable to handleAuthResponse"; log.warn("[{}] {} ", remoteAddress, msg, e); - ctx.writeAndFlush(Commands.newError(-1, ServerError.UnknownError, msg)); + writeAndFlushVoidPromise(Commands.newError(-1, ServerError.UnknownError, msg)); close(); } } @@ -1211,7 +1213,7 @@ protected void handleSubscribe(final CommandSubscribe subscribe) { String msg = "Client is not authorized to subscribe"; log.warn("[{}] {} with role {}", remoteAddress, msg, getPrincipal()); consumers.remove(consumerId, consumerFuture); - ctx.writeAndFlush(Commands.newError(requestId, ServerError.AuthorizationError, msg)); + writeAndFlushVoidPromise(Commands.newError(requestId, ServerError.AuthorizationError, msg)); } return null; }).exceptionally(ex -> { @@ -1288,7 +1290,7 @@ protected void handleProducer(final CommandProducer cmdProducer) { if (!isAuthorized) { String msg = "Client is not authorized to Produce"; log.warn("[{}] {} with role {}", remoteAddress, msg, getPrincipal()); - ctx.writeAndFlush(Commands.newError(requestId, ServerError.AuthorizationError, msg)); + writeAndFlushVoidPromise(Commands.newError(requestId, ServerError.AuthorizationError, msg)); return null; } @@ -1658,7 +1660,7 @@ protected void handleAck(CommandAck ack) { Consumer consumer = consumerFuture.getNow(null); consumer.messageAcked(ack).thenRun(() -> { if (hasRequestId) { - ctx.writeAndFlush(Commands.newAckResponse( + writeAndFlushVoidPromise(Commands.newAckResponse( requestId, null, null, consumerId)); } if (brokerInterceptor != null) { @@ -1666,7 +1668,7 @@ protected void handleAck(CommandAck ack) { } }).exceptionally(e -> { if (hasRequestId) { - ctx.writeAndFlush(Commands.newAckResponse(requestId, + writeAndFlushVoidPromise(Commands.newAckResponse(requestId, BrokerServiceException.getClientErrorCode(e), e.getMessage(), consumerId)); } @@ -1825,7 +1827,7 @@ protected void handleCloseProducer(CommandCloseProducer closeProducer) { CompletableFuture producerFuture = producers.get(producerId); if (producerFuture == null) { log.info("[{}] Producer {} was not registered on the connection", remoteAddress, producerId); - ctx.writeAndFlush(Commands.newSuccess(requestId)); + writeAndFlushVoidPromise(Commands.newSuccess(requestId)); return; } @@ -1874,7 +1876,7 @@ protected void handleCloseConsumer(CommandCloseConsumer closeConsumer) { CompletableFuture consumerFuture = consumers.get(consumerId); if (consumerFuture == null) { log.info("[{}] Consumer was not registered on the connection: {}", consumerId, remoteAddress); - ctx.writeAndFlush(Commands.newSuccess(requestId)); + writeAndFlushVoidPromise(Commands.newSuccess(requestId)); return; } @@ -1941,7 +1943,7 @@ protected void handleGetLastMessageId(CommandGetLastMessageId getLastMessageId) consumer.getSubscription().getName()); } else { - ctx.writeAndFlush(Commands.newError(getLastMessageId.getRequestId(), + writeAndFlushVoidPromise(Commands.newError(getLastMessageId.getRequestId(), ServerError.MetadataError, "Consumer not found")); } } @@ -1994,7 +1996,7 @@ public void readEntryFailed(ManagedLedgerException exception, Object ctx) { handleLastMessageIdFromCompactedLedger(persistentTopic, requestId, partitionIndex, markDeletePosition); } else { - ctx.writeAndFlush(Commands.newError( + writeAndFlushVoidPromise(Commands.newError( requestId, ServerError.MetadataError, "Failed to get batch size for entry " + e.getMessage())); } @@ -2006,7 +2008,7 @@ public void readEntryFailed(ManagedLedgerException exception, Object ctx) { topic.getName(), subscriptionName, lastPosition, partitionIndex); } - ctx.writeAndFlush(Commands.newGetLastMessageIdResponse(requestId, lastPosition.getLedgerId(), + writeAndFlushVoidPromise(Commands.newGetLastMessageIdResponse(requestId, lastPosition.getLedgerId(), lastPosition.getEntryId(), partitionIndex, largestBatchIndex, markDeletePosition != null ? markDeletePosition.getLedgerId() : -1, markDeletePosition != null ? markDeletePosition.getEntryId() : -1)); @@ -2026,12 +2028,12 @@ private void handleLastMessageIdFromCompactedLedger(PersistentTopic persistentTo try { largestBatchIndex = calculateTheLastBatchIndexInBatch(metadata, payload); } catch (IOException ioEx){ - ctx.writeAndFlush(Commands.newError(requestId, ServerError.MetadataError, + writeAndFlushVoidPromise(Commands.newError(requestId, ServerError.MetadataError, "Failed to deserialize batched message from the last entry of the compacted Ledger: " + ioEx.getMessage())); return; } - ctx.writeAndFlush(Commands.newGetLastMessageIdResponse(requestId, + writeAndFlushVoidPromise(Commands.newGetLastMessageIdResponse(requestId, entry.getLedgerId(), entry.getEntryId(), partitionIndex, largestBatchIndex, markDeletePosition != null ? markDeletePosition.getLedgerId() : -1, markDeletePosition != null ? markDeletePosition.getEntryId() : -1)); @@ -2039,13 +2041,13 @@ private void handleLastMessageIdFromCompactedLedger(PersistentTopic persistentTo } else { // in this case, the ledgers been removed except the current ledger // and current ledger without any data - ctx.writeAndFlush(Commands.newGetLastMessageIdResponse(requestId, + writeAndFlushVoidPromise(Commands.newGetLastMessageIdResponse(requestId, -1, -1, partitionIndex, -1, markDeletePosition != null ? markDeletePosition.getLedgerId() : -1, markDeletePosition != null ? markDeletePosition.getEntryId() : -1)); } }).exceptionally(ex -> { - ctx.writeAndFlush(Commands.newError( + writeAndFlushVoidPromise(Commands.newError( requestId, ServerError.MetadataError, "Failed to read last entry of the compacted Ledger " + ex.getCause().getMessage())); @@ -2401,12 +2403,12 @@ protected void handleAddPartitionToTxn(CommandAddPartitionToTxn command) { if (log.isDebugEnabled()) { log.debug("Send response success for add published partition to txn request {}", requestId); } - ctx.writeAndFlush(Commands.newAddPartitionToTxnResponse(requestId, + writeAndFlushVoidPromise(Commands.newAddPartitionToTxnResponse(requestId, txnID.getLeastSigBits(), txnID.getMostSigBits())); } else { ex = handleTxnException(ex, BaseCommand.Type.ADD_PARTITION_TO_TXN.name(), requestId); - ctx.writeAndFlush(Commands.newAddPartitionToTxnResponse(requestId, + writeAndFlushVoidPromise(Commands.newAddPartitionToTxnResponse(requestId, txnID.getLeastSigBits(), txnID.getMostSigBits(), BrokerServiceException.getClientErrorCode(ex), @@ -2468,13 +2470,13 @@ protected void handleEndTxnOnPartition(CommandEndTxnOnPartition command) { if (throwable != null) { log.error("handleEndTxnOnPartition fail!, topic {}, txnId: [{}], " + "txnAction: [{}]", topic, txnID, TxnAction.valueOf(txnAction), throwable); - ctx.writeAndFlush(Commands.newEndTxnOnPartitionResponse( + writeAndFlushVoidPromise(Commands.newEndTxnOnPartitionResponse( requestId, BrokerServiceException.getClientErrorCode(throwable), throwable.getMessage(), txnID.getLeastSigBits(), txnID.getMostSigBits())); return; } - ctx.writeAndFlush(Commands.newEndTxnOnPartitionResponse(requestId, + writeAndFlushVoidPromise(Commands.newEndTxnOnPartitionResponse(requestId, txnID.getLeastSigBits(), txnID.getMostSigBits())); }); @@ -2486,7 +2488,7 @@ protected void handleEndTxnOnPartition(CommandEndTxnOnPartition command) { log.error("handleEndTxnOnPartition fail ! The topic {} does not exist in broker, " + "txnId: [{}], txnAction: [{}]", topic, txnID, TxnAction.valueOf(txnAction)); - ctx.writeAndFlush(Commands.newEndTxnOnPartitionResponse(requestId, + writeAndFlushVoidPromise(Commands.newEndTxnOnPartitionResponse(requestId, ServerError.ServiceNotReady, "The topic " + topic + " does not exist in broker.", txnID.getLeastSigBits(), txnID.getMostSigBits())); @@ -2494,14 +2496,14 @@ protected void handleEndTxnOnPartition(CommandEndTxnOnPartition command) { log.warn("handleEndTxnOnPartition fail ! The topic {} has not been created, " + "txnId: [{}], txnAction: [{}]", topic, txnID, TxnAction.valueOf(txnAction)); - ctx.writeAndFlush(Commands.newEndTxnOnPartitionResponse(requestId, + writeAndFlushVoidPromise(Commands.newEndTxnOnPartitionResponse(requestId, txnID.getLeastSigBits(), txnID.getMostSigBits())); } }).exceptionally(e -> { log.error("handleEndTxnOnPartition fail ! topic {}, " + "txnId: [{}], txnAction: [{}]", topic, txnID, TxnAction.valueOf(txnAction), e.getCause()); - ctx.writeAndFlush(Commands.newEndTxnOnPartitionResponse( + writeAndFlushVoidPromise(Commands.newEndTxnOnPartitionResponse( requestId, ServerError.ServiceNotReady, e.getMessage(), txnID.getLeastSigBits(), txnID.getMostSigBits())); return null; @@ -2511,7 +2513,7 @@ protected void handleEndTxnOnPartition(CommandEndTxnOnPartition command) { log.error("handleEndTxnOnPartition fail ! topic {}, " + "txnId: [{}], txnAction: [{}]", topic, txnID, TxnAction.valueOf(txnAction), e.getCause()); - ctx.writeAndFlush(Commands.newEndTxnOnPartitionResponse( + writeAndFlushVoidPromise(Commands.newEndTxnOnPartitionResponse( requestId, ServerError.ServiceNotReady, e.getMessage(), txnID.getLeastSigBits(), txnID.getMostSigBits())); return null; @@ -2543,8 +2545,7 @@ protected void handleEndTxnOnSubscription(CommandEndTxnOnSubscription command) { log.warn("handleEndTxnOnSubscription fail! " + "topic {} subscription {} does not exist. txnId: [{}], txnAction: [{}]", optionalTopic.get().getName(), subName, txnID, TxnAction.valueOf(txnAction)); - ctx.writeAndFlush( - Commands.newEndTxnOnSubscriptionResponse(requestId, txnidLeastBits, txnidMostBits)); + writeAndFlushVoidPromise(Commands.newEndTxnOnSubscriptionResponse(requestId, txnidLeastBits, txnidMostBits)); return; } @@ -2555,14 +2556,13 @@ protected void handleEndTxnOnSubscription(CommandEndTxnOnSubscription command) { log.error("handleEndTxnOnSubscription fail ! topic: {}, subscription: {}" + "txnId: [{}], txnAction: [{}]", topic, subName, txnID, TxnAction.valueOf(txnAction), e.getCause()); - ctx.writeAndFlush(Commands.newEndTxnOnSubscriptionResponse( + writeAndFlushVoidPromise(Commands.newEndTxnOnSubscriptionResponse( requestId, txnidLeastBits, txnidMostBits, BrokerServiceException.getClientErrorCode(e), "Handle end txn on subscription failed: " + e.getMessage())); return; } - ctx.writeAndFlush( - Commands.newEndTxnOnSubscriptionResponse(requestId, txnidLeastBits, txnidMostBits)); + writeAndFlushVoidPromise(Commands.newEndTxnOnSubscriptionResponse(requestId, txnidLeastBits, txnidMostBits)); }); } else { getBrokerService().getManagedLedgerFactory() @@ -2572,7 +2572,7 @@ protected void handleEndTxnOnSubscription(CommandEndTxnOnSubscription command) { log.error("handleEndTxnOnSubscription fail! The topic {} does not exist in broker, " + "subscription: {}, txnId: [{}], txnAction: [{}]", topic, subName, txnID, TxnAction.valueOf(txnAction)); - ctx.writeAndFlush(Commands.newEndTxnOnSubscriptionResponse( + writeAndFlushVoidPromise(Commands.newEndTxnOnSubscriptionResponse( requestId, txnID.getLeastSigBits(), txnID.getMostSigBits(), ServerError.ServiceNotReady, "The topic " + topic + " does not exist in broker.")); @@ -2580,14 +2580,14 @@ protected void handleEndTxnOnSubscription(CommandEndTxnOnSubscription command) { log.warn("handleEndTxnOnSubscription fail ! The topic {} has not been created, " + "subscription: {} txnId: [{}], txnAction: [{}]", topic, subName, txnID, TxnAction.valueOf(txnAction)); - ctx.writeAndFlush(Commands.newEndTxnOnSubscriptionResponse(requestId, + writeAndFlushVoidPromise(Commands.newEndTxnOnSubscriptionResponse(requestId, txnID.getLeastSigBits(), txnID.getMostSigBits())); } }).exceptionally(e -> { log.error("handleEndTxnOnSubscription fail ! topic {}, subscription: {}" + "txnId: [{}], txnAction: [{}]", topic, subName, txnID, TxnAction.valueOf(txnAction), e.getCause()); - ctx.writeAndFlush(Commands.newEndTxnOnSubscriptionResponse( + writeAndFlushVoidPromise(Commands.newEndTxnOnSubscriptionResponse( requestId, txnID.getLeastSigBits(), txnID.getMostSigBits(), ServerError.ServiceNotReady, e.getMessage())); return null; @@ -2597,7 +2597,7 @@ protected void handleEndTxnOnSubscription(CommandEndTxnOnSubscription command) { log.error("handleEndTxnOnSubscription fail ! topic: {}, subscription: {}" + "txnId: [{}], txnAction: [{}]", topic, subName, txnID, TxnAction.valueOf(txnAction), e.getCause()); - ctx.writeAndFlush(Commands.newEndTxnOnSubscriptionResponse( + writeAndFlushVoidPromise(Commands.newEndTxnOnSubscriptionResponse( requestId, txnidLeastBits, txnidMostBits, ServerError.ServiceNotReady, "Handle end txn on subscription failed: " + e.getMessage())); @@ -2652,12 +2652,12 @@ protected void handleAddSubscriptionToTxn(CommandAddSubscriptionToTxn command) { log.debug("Send response success for add published partition to txn request {}", requestId); } - ctx.writeAndFlush(Commands.newAddSubscriptionToTxnResponse(requestId, + writeAndFlushVoidPromise(Commands.newAddSubscriptionToTxnResponse(requestId, txnID.getLeastSigBits(), txnID.getMostSigBits())); } else { ex = handleTxnException(ex, BaseCommand.Type.ADD_SUBSCRIPTION_TO_TXN.name(), requestId); - - ctx.writeAndFlush(Commands.newAddSubscriptionToTxnResponse(requestId, txnID.getLeastSigBits(), + writeAndFlushVoidPromise( + Commands.newAddSubscriptionToTxnResponse(requestId, txnID.getLeastSigBits(), txnID.getMostSigBits(), BrokerServiceException.getClientErrorCode(ex), ex.getMessage())); transactionMetadataStoreService.handleOpFail(ex, tcId); @@ -2743,7 +2743,7 @@ public void closeProducer(Producer producer) { // removes producer-connection from map and send close command to producer safelyRemoveProducer(producer); if (getRemoteEndpointProtocolVersion() >= v5.getValue()) { - ctx.writeAndFlush(Commands.newCloseProducer(producer.getProducerId(), -1L)); + writeAndFlushVoidPromise(Commands.newCloseProducer(producer.getProducerId(), -1L)); } else { close(); } @@ -2755,7 +2755,7 @@ public void closeConsumer(Consumer consumer) { // removes consumer-connection from map and send close command to consumer safelyRemoveConsumer(consumer); if (getRemoteEndpointProtocolVersion() >= v5.getValue()) { - ctx.writeAndFlush(Commands.newCloseConsumer(consumer.consumerId(), -1L)); + writeAndFlushVoidPromise(Commands.newCloseConsumer(consumer.consumerId(), -1L)); } else { close(); } @@ -2998,13 +2998,13 @@ private TopicName validateTopicName(String topic, long requestId, Object request } if (requestCommand instanceof CommandLookupTopic) { - ctx.writeAndFlush(Commands.newLookupErrorResponse(ServerError.InvalidTopicName, + writeAndFlushVoidPromise(Commands.newLookupErrorResponse(ServerError.InvalidTopicName, "Invalid topic name: " + t.getMessage(), requestId)); } else if (requestCommand instanceof CommandPartitionedTopicMetadata) { - ctx.writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.InvalidTopicName, + writeAndFlushVoidPromise(Commands.newPartitionMetadataResponse(ServerError.InvalidTopicName, "Invalid topic name: " + t.getMessage(), requestId)); } else { - ctx.writeAndFlush(Commands.newError(requestId, ServerError.InvalidTopicName, + writeAndFlushVoidPromise(Commands.newError(requestId, ServerError.InvalidTopicName, "Invalid topic name: " + t.getMessage())); } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java index 72212fe16c7e4..e122e557c491d 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java @@ -22,6 +22,7 @@ import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.ChannelOutboundInvoker; import io.netty.handler.codec.haproxy.HAProxyMessage; import org.apache.pulsar.common.api.proto.BaseCommand; import org.apache.pulsar.common.api.proto.CommandAck; @@ -131,7 +132,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception interceptCommand(cmd); handlePartitionMetadataRequest(cmd.getPartitionMetadata()); } catch (InterceptException e) { - ctx.writeAndFlush(Commands.newPartitionMetadataResponse(getServerError(e.getErrorCode()), + writeAndFlushVoidPromise(ctx, Commands.newPartitionMetadataResponse(getServerError(e.getErrorCode()), e.getMessage(), cmd.getPartitionMetadata().getRequestId())); } break; @@ -205,7 +206,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception interceptCommand(cmd); handleProducer(cmd.getProducer()); } catch (InterceptException e) { - ctx.writeAndFlush(Commands.newError(cmd.getProducer().getRequestId(), + writeAndFlushVoidPromise(ctx, Commands.newError(cmd.getProducer().getRequestId(), getServerError(e.getErrorCode()), e.getMessage())); } break; @@ -218,7 +219,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception ByteBuf headersAndPayload = buffer.markReaderIndex(); handleSend(cmd.getSend(), headersAndPayload); } catch (InterceptException e) { - ctx.writeAndFlush(Commands.newSendError(cmd.getSend().getProducerId(), + writeAndFlushVoidPromise(ctx, Commands.newSendError(cmd.getSend().getProducerId(), cmd.getSend().getSequenceId(), getServerError(e.getErrorCode()), e.getMessage())); } break; @@ -239,7 +240,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception interceptCommand(cmd); handleSubscribe(cmd.getSubscribe()); } catch (InterceptException e) { - ctx.writeAndFlush(Commands.newError(cmd.getSubscribe().getRequestId(), + writeAndFlushVoidPromise(ctx, Commands.newError(cmd.getSubscribe().getRequestId(), getServerError(e.getErrorCode()), e.getMessage())); } break; @@ -266,7 +267,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception interceptCommand(cmd); handleSeek(cmd.getSeek()); } catch (InterceptException e) { - ctx.writeAndFlush(Commands.newError(cmd.getSeek().getRequestId(), getServerError(e.getErrorCode()), + writeAndFlushVoidPromise(ctx, Commands.newError(cmd.getSeek().getRequestId(), getServerError(e.getErrorCode()), e.getMessage())); } break; @@ -326,7 +327,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception interceptCommand(cmd); handleGetTopicsOfNamespace(cmd.getGetTopicsOfNamespace()); } catch (InterceptException e) { - ctx.writeAndFlush(Commands.newError(cmd.getGetTopicsOfNamespace().getRequestId(), + writeAndFlushVoidPromise(ctx, Commands.newError(cmd.getGetTopicsOfNamespace().getRequestId(), getServerError(e.getErrorCode()), e.getMessage())); } break; @@ -342,7 +343,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception interceptCommand(cmd); handleGetSchema(cmd.getGetSchema()); } catch (InterceptException e) { - ctx.writeAndFlush(Commands.newGetSchemaResponseError(cmd.getGetSchema().getRequestId(), + writeAndFlushVoidPromise(ctx, Commands.newGetSchemaResponseError(cmd.getGetSchema().getRequestId(), getServerError(e.getErrorCode()), e.getMessage())); } break; @@ -358,7 +359,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception interceptCommand(cmd); handleGetOrCreateSchema(cmd.getGetOrCreateSchema()); } catch (InterceptException e) { - ctx.writeAndFlush(Commands.newGetOrCreateSchemaResponseError( + writeAndFlushVoidPromise(ctx, Commands.newGetOrCreateSchemaResponseError( cmd.getGetOrCreateSchema().getRequestId(), getServerError(e.getErrorCode()), e.getMessage())); } @@ -731,4 +732,8 @@ protected void handleCommandWatchTopicListClose(CommandWatchTopicListClose comma } private static final Logger log = LoggerFactory.getLogger(PulsarDecoder.class); + + private void writeAndFlushVoidPromise(ChannelOutboundInvoker ctx, ByteBuf cmd) { + ctx.writeAndFlush(cmd, ctx.voidPromise()); + } } diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java index 5101978c3e2dc..662afb18d1858 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java @@ -19,6 +19,7 @@ package org.apache.pulsar.proxy.server; import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelOutboundInvoker; import io.prometheus.client.Counter; import java.net.InetSocketAddress; import java.net.SocketAddress; @@ -113,7 +114,7 @@ public void handleLookup(CommandLookupTopic lookup) { log.debug("Lookup Request ID {} from {} rejected - {}.", clientRequestId, clientAddress, throttlingErrorMessage); } - proxyConnection.ctx().writeAndFlush(Commands.newLookupErrorResponse(ServerError.ServiceNotReady, + writeAndFlushVoidPromise(Commands.newLookupErrorResponse(ServerError.ServiceNotReady, throttlingErrorMessage, clientRequestId)); } @@ -122,7 +123,7 @@ public void handleLookup(CommandLookupTopic lookup) { private void performLookup(long clientRequestId, String topic, String brokerServiceUrl, boolean authoritative, int numberOfRetries) { if (numberOfRetries == 0) { - proxyConnection.ctx().writeAndFlush(Commands.newLookupErrorResponse(ServerError.ServiceNotReady, + writeAndFlushVoidPromise(Commands.newLookupErrorResponse(ServerError.ServiceNotReady, "Reached max number of redirections", clientRequestId)); return; } @@ -131,7 +132,7 @@ private void performLookup(long clientRequestId, String topic, String brokerServ try { brokerURI = new URI(brokerServiceUrl); } catch (URISyntaxException e) { - proxyConnection.ctx().writeAndFlush( + writeAndFlushVoidPromise( Commands.newLookupErrorResponse(ServerError.MetadataError, e.getMessage(), clientRequestId)); return; } @@ -150,7 +151,7 @@ private void performLookup(long clientRequestId, String topic, String brokerServ clientCnx.newLookup(command, requestId).whenComplete((r, t) -> { if (t != null) { log.warn("[{}] Failed to lookup topic {}: {}", clientAddress, topic, t.getMessage()); - proxyConnection.ctx().writeAndFlush( + writeAndFlushVoidPromise( Commands.newLookupErrorResponse(getServerError(t), t.getMessage(), clientRequestId)); } else { String brokerUrl = connectWithTLS ? r.brokerUrlTls : r.brokerUrl; @@ -170,7 +171,7 @@ private void performLookup(long clientRequestId, String topic, String brokerServ + " with clientReq Id '{}' and lookup-broker {}", addr, topic, clientRequestId, brokerUrl); } - proxyConnection.ctx().writeAndFlush(Commands.newLookupResponse(brokerUrl, brokerUrl, true, + writeAndFlushVoidPromise(Commands.newLookupResponse(brokerUrl, brokerUrl, true, LookupType.Connect, clientRequestId, true /* this is coming from proxy */)); } } @@ -178,7 +179,7 @@ private void performLookup(long clientRequestId, String topic, String brokerServ }); }).exceptionally(ex -> { // Failed to connect to backend broker - proxyConnection.ctx().writeAndFlush( + writeAndFlushVoidPromise( Commands.newLookupErrorResponse(getServerError(ex), ex.getMessage(), clientRequestId)); return null; }); @@ -202,7 +203,7 @@ public void handlePartitionMetadataResponse(CommandPartitionedTopicMetadata part log.debug("PartitionMetaData Request ID {} from {} rejected - {}.", clientRequestId, clientAddress, throttlingErrorMessage); } - proxyConnection.ctx().writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.ServiceNotReady, + writeAndFlushVoidPromise(Commands.newPartitionMetadataResponse(ServerError.ServiceNotReady, throttlingErrorMessage, clientRequestId)); } } @@ -239,17 +240,17 @@ private void handlePartitionMetadataResponse(CommandPartitionedTopicMetadata par if (t != null) { log.warn("[{}] failed to get Partitioned metadata : {}", topicName.toString(), t.getMessage(), t); - proxyConnection.ctx().writeAndFlush(Commands.newLookupErrorResponse(getServerError(t), + writeAndFlushVoidPromise(Commands.newLookupErrorResponse(getServerError(t), t.getMessage(), clientRequestId)); } else { - proxyConnection.ctx().writeAndFlush( + writeAndFlushVoidPromise( Commands.newPartitionMetadataResponse(r.partitions, clientRequestId)); } proxyConnection.getConnectionPool().releaseConnection(clientCnx); }); }).exceptionally(ex -> { // Failed to connect to backend broker - proxyConnection.ctx().writeAndFlush(Commands.newPartitionMetadataResponse(getServerError(ex), + writeAndFlushVoidPromise(Commands.newPartitionMetadataResponse(getServerError(ex), ex.getMessage(), clientRequestId)); return null; }); @@ -275,7 +276,7 @@ public void handleGetTopicsOfNamespace(CommandGetTopicsOfNamespace commandGetTop log.debug("GetTopicsOfNamespace Request ID {} from {} rejected - {}.", requestId, clientAddress, throttlingErrorMessage); } - proxyConnection.ctx().writeAndFlush(Commands.newError( + writeAndFlushVoidPromise(Commands.newError( requestId, ServerError.ServiceNotReady, throttlingErrorMessage )); } @@ -304,7 +305,7 @@ private void performGetTopicsOfNamespace(long clientRequestId, String topicsHash, CommandGetTopicsOfNamespace.Mode mode) { if (numberOfRetries == 0) { - proxyConnection.ctx().writeAndFlush(Commands.newError(clientRequestId, ServerError.ServiceNotReady, + writeAndFlushVoidPromise(Commands.newError(clientRequestId, ServerError.ServiceNotReady, "Reached max number of redirections")); return; } @@ -329,10 +330,10 @@ private void performGetTopicsOfNamespace(long clientRequestId, if (t != null) { log.warn("[{}] Failed to get TopicsOfNamespace {}: {}", clientAddress, namespaceName, t.getMessage()); - proxyConnection.ctx().writeAndFlush( + writeAndFlushVoidPromise( Commands.newError(clientRequestId, getServerError(t), t.getMessage())); } else { - proxyConnection.ctx().writeAndFlush( + writeAndFlushVoidPromise( Commands.newGetTopicsOfNamespaceResponse(r.getTopics(), r.getTopicsHash(), r.isFiltered(), r.isChanged(), clientRequestId)); } @@ -341,7 +342,7 @@ private void performGetTopicsOfNamespace(long clientRequestId, proxyConnection.getConnectionPool().releaseConnection(clientCnx); }).exceptionally(ex -> { // Failed to connect to backend broker - proxyConnection.ctx().writeAndFlush( + writeAndFlushVoidPromise( Commands.newError(clientRequestId, getServerError(ex), ex.getMessage())); return null; }); @@ -384,10 +385,10 @@ public void handleGetSchema(CommandGetSchema commandGetSchema) { clientCnx.sendGetRawSchema(command, requestId).whenComplete((r, t) -> { if (t != null) { log.warn("[{}] Failed to get schema {}: {}", clientAddress, topic, t); - proxyConnection.ctx().writeAndFlush( + writeAndFlushVoidPromise( Commands.newError(clientRequestId, getServerError(t), t.getMessage())); } else { - proxyConnection.ctx().writeAndFlush( + writeAndFlushVoidPromise( Commands.newGetSchemaResponse(clientRequestId, r)); } @@ -395,7 +396,7 @@ public void handleGetSchema(CommandGetSchema commandGetSchema) { }); }).exceptionally(ex -> { // Failed to connect to backend broker - proxyConnection.ctx().writeAndFlush( + writeAndFlushVoidPromise( Commands.newError(clientRequestId, getServerError(ex), ex.getMessage())); return null; }); @@ -414,7 +415,7 @@ private String getBrokerServiceUrl(long clientRequestId) { availableBroker = discoveryProvider.nextBroker(); } catch (Exception e) { log.warn("[{}] Failed to get next active broker {}", clientAddress, e.getMessage(), e); - proxyConnection.ctx().writeAndFlush(Commands.newError( + writeAndFlushVoidPromise(Commands.newError( clientRequestId, ServerError.ServiceNotReady, e.getMessage() )); return null; @@ -427,7 +428,7 @@ private InetSocketAddress getAddr(String brokerServiceUrl, long clientRequestId) try { brokerURI = new URI(brokerServiceUrl); } catch (URISyntaxException e) { - proxyConnection.ctx().writeAndFlush( + writeAndFlushVoidPromise( Commands.newError(clientRequestId, ServerError.MetadataError, e.getMessage())); return null; } @@ -446,5 +447,9 @@ private ServerError getServerError(Throwable error) { return responseError; } + private void writeAndFlushVoidPromise(ByteBuf cmd) { + proxyConnection.ctx().writeAndFlush(cmd); + } + private static final Logger log = LoggerFactory.getLogger(LookupProxyHandler.class); } From f239270ba3d289ffc351d4e9b933c7103a5360e8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=B2=20Boschi?= Date: Thu, 5 Jan 2023 16:58:26 +0100 Subject: [PATCH 2/4] =?UTF-8?q?[improve][broker,proxy]=C2=A0Use=20ChannelV?= =?UTF-8?q?oidPromise=20to=20avoid=20useless=20promise=20objects=20creatio?= =?UTF-8?q?n?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/PulsarCommandSenderImpl.java | 55 ++++---- .../pulsar/broker/service/ServerCnx.java | 129 ++++++++++-------- .../pulsar/common/protocol/PulsarDecoder.java | 29 ++-- .../common/util/netty/NettyChannelUtil.java | 38 ++++++ .../util/netty/NettyChannelUtilTest.java | 68 +++++++++ .../proxy/server/DirectProxyHandler.java | 30 ++-- .../proxy/server/LookupProxyHandler.java | 48 +++---- .../pulsar/proxy/server/ProxyConnection.java | 54 ++++---- 8 files changed, 292 insertions(+), 159 deletions(-) create mode 100644 pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/NettyChannelUtil.java create mode 100644 pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/NettyChannelUtilTest.java 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 0b489f42e5d5f..6510da1fbe72a 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 @@ -38,6 +38,7 @@ import org.apache.pulsar.common.protocol.Commands; import org.apache.pulsar.common.protocol.schema.SchemaVersion; import org.apache.pulsar.common.schema.SchemaInfo; +import org.apache.pulsar.common.util.netty.NettyChannelUtil; @Slf4j public class PulsarCommandSenderImpl implements PulsarCommandSender { @@ -55,7 +56,7 @@ public void sendPartitionMetadataResponse(ServerError error, String errorMsg, lo BaseCommand command = Commands.newPartitionMetadataResponseCommand(error, errorMsg, requestId); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -63,7 +64,7 @@ public void sendPartitionMetadataResponse(int partitions, long requestId) { BaseCommand command = Commands.newPartitionMetadataResponseCommand(partitions, requestId); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -71,7 +72,7 @@ public void sendSuccessResponse(long requestId) { BaseCommand command = Commands.newSuccessCommand(requestId); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -79,7 +80,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); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -87,7 +88,7 @@ public void sendProducerSuccessResponse(long requestId, String producerName, Sch BaseCommand command = Commands.newProducerSuccessCommand(requestId, producerName, schemaVersion); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -98,7 +99,7 @@ public void sendProducerSuccessResponse(long requestId, String producerName, lon schemaVersion, topicEpoch, isProducerReady); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -108,7 +109,7 @@ public void sendSendReceiptResponse(long producerId, long sequenceId, long highe entryId); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -116,7 +117,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); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -126,7 +127,7 @@ public void sendGetTopicsOfNamespaceResponse(List topics, String topicsH filtered, changed, requestId); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -134,7 +135,7 @@ public void sendGetSchemaResponse(long requestId, SchemaInfo schema, SchemaVersi BaseCommand command = Commands.newGetSchemaResponseCommand(requestId, schema, version); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -142,7 +143,7 @@ public void sendGetSchemaErrorResponse(long requestId, ServerError error, String BaseCommand command = Commands.newGetSchemaResponseErrorCommand(requestId, error, errorMessage); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -150,7 +151,7 @@ public void sendGetOrCreateSchemaResponse(long requestId, SchemaVersion schemaVe BaseCommand command = Commands.newGetOrCreateSchemaResponseCommand(requestId, schemaVersion); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -159,7 +160,7 @@ public void sendGetOrCreateSchemaErrorResponse(long requestId, ServerError error Commands.newGetOrCreateSchemaResponseErrorCommand(requestId, error, errorMessage); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -168,7 +169,7 @@ public void sendConnectedResponse(int clientProtocolVersion, int maxMessageSize, clientProtocolVersion, maxMessageSize, supportsTopicWatchers); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -179,7 +180,7 @@ public void sendLookupResponse(String brokerServiceUrl, String brokerServiceUrlT authoritative, response, requestId, proxyThroughServiceUrl); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -187,7 +188,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); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -196,7 +197,7 @@ public void sendActiveConsumerChange(long consumerId, boolean isActive) { // if the client is older than `v12`, we don't need to send consumer group changes. return; } - writeAndFlushVoidPromise(Commands.newActiveConsumerChange(consumerId, isActive)); + writeAndFlush(Commands.newActiveConsumerChange(consumerId, isActive)); } @Override @@ -204,7 +205,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); - writeAndFlushVoidPromise(Commands.newReachedEndOfTopic(consumerId)); + writeAndFlush(Commands.newReachedEndOfTopic(consumerId)); } } @@ -213,7 +214,7 @@ public boolean sendTopicMigrated(ResourceType type, long resourceId, String brok // Only send notification if the client understand the command if (cnx.getRemoteEndpointProtocolVersion() >= ProtocolVersion.v20.getValue()) { log.info("[{}] Notifying {} that topic is migrated", type.name(), resourceId); - writeAndFlushVoidPromise(Commands.newTopicMigrated(type, resourceId, brokerUrl, brokerUrlTls)); + writeAndFlush(Commands.newTopicMigrated(type, resourceId, brokerUrl, brokerUrlTls)); return true; } return false; @@ -307,7 +308,7 @@ public void sendTcClientConnectResponse(long requestId, ServerError error, Strin BaseCommand command = Commands.newTcClientConnectResponse(requestId, error, message); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -321,7 +322,7 @@ public void sendNewTxnResponse(long requestId, TxnID txnID, long tcID) { txnID.getMostSigBits()); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); if (this.interceptor != null) { this.interceptor.txnOpened(tcID, txnID.toString()); } @@ -332,7 +333,7 @@ public void sendNewTxnErrorResponse(long requestId, long tcID, ServerError error BaseCommand command = Commands.newTxnResponse(requestId, tcID, error, message); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } @Override @@ -341,7 +342,7 @@ public void sendEndTxnResponse(long requestId, TxnID txnID, int txnAction) { txnID.getMostSigBits()); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); if (this.interceptor != null) { this.interceptor.txnEnded(txnID.toString(), txnAction); } @@ -353,7 +354,7 @@ public void sendEndTxnErrorResponse(long requestId, TxnID txnID, ServerError err txnID.getMostSigBits(), error, message); safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); if (this.interceptor != null) { this.interceptor.txnEnded(txnID.toString(), TxnAction.ABORT_VALUE); } @@ -375,11 +376,11 @@ public void sendWatchTopicListUpdate(long watcherId, private void interceptAndWriteCommand(BaseCommand command) { safeIntercept(command, cnx); ByteBuf outBuf = Commands.serializeWithSize(command); - writeAndFlushVoidPromise(outBuf); + writeAndFlush(outBuf); } - private void writeAndFlushVoidPromise(ByteBuf outBuf) { - cnx.ctx().writeAndFlush(outBuf, cnx.ctx().voidPromise()); + private void writeAndFlush(ByteBuf outBuf) { + NettyChannelUtil.writeAndFlushWithVoidPromise(cnx.ctx(), outBuf); } private void safeIntercept(BaseCommand command, ServerCnx cnx) { 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 49b125e069247..aebb55d097e9a 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 @@ -161,6 +161,7 @@ import org.apache.pulsar.common.topics.TopicList; import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.common.util.collections.ConcurrentLongHashMap; +import org.apache.pulsar.common.util.netty.NettyChannelUtil; import org.apache.pulsar.functions.utils.Exceptions; import org.apache.pulsar.transaction.coordinator.TransactionCoordinatorID; import org.apache.pulsar.transaction.coordinator.exceptions.CoordinatorException; @@ -304,11 +305,11 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception { super.channelActive(ctx); ConnectionController.State state = connectionController.increaseConnection(remoteAddress); if (!state.equals(ConnectionController.State.OK)) { - ctx.writeAndFlush(Commands.newError(-1, ServerError.NotAllowedError, - state.equals(ConnectionController.State.REACH_MAX_CONNECTION) - ? "Reached the maximum number of connections" - : "Reached the maximum number of connections on address" + remoteAddress)) - .addListener(ChannelFutureListener.CLOSE); + final ByteBuf msg = Commands.newError(-1, ServerError.NotAllowedError, + state.equals(ConnectionController.State.REACH_MAX_CONNECTION) + ? "Reached the maximum number of connections" + : "Reached the maximum number of connections on address" + remoteAddress); + NettyChannelUtil.writeAndFlushWithClosePromise(ctx, msg); return; } log.info("New connection from {}", remoteAddress); @@ -480,7 +481,7 @@ protected void handleLookup(CommandLookupTopic lookup) { log.debug("[{}] Failed lookup topic {} due to pulsar service is not ready: {} state", remoteAddress, topicName, this.service.getPulsar().getState().toString()); } - writeAndFlushVoidPromise(newLookupErrorResponse(ServerError.ServiceNotReady, + writeAndFlush(newLookupErrorResponse(ServerError.ServiceNotReady, "Failed due to pulsar service is not ready", requestId)); return; } @@ -491,7 +492,7 @@ protected void handleLookup(CommandLookupTopic lookup) { final String msg = "Valid Proxy Client role should be provided for lookup "; log.warn("[{}] {} with role {} and proxyClientAuthRole {} on topic {}", remoteAddress, msg, authRole, originalPrincipal, topicName); - writeAndFlushVoidPromise(newLookupErrorResponse(ServerError.AuthorizationError, msg, requestId)); + writeAndFlush(newLookupErrorResponse(ServerError.AuthorizationError, msg, requestId)); lookupSemaphore.release(); return; } @@ -502,12 +503,12 @@ protected void handleLookup(CommandLookupTopic lookup) { getPrincipal(), getAuthenticationData(), requestId, advertisedListenerName).handle((lookupResponse, ex) -> { if (ex == null) { - writeAndFlushVoidPromise(lookupResponse); + writeAndFlush(lookupResponse); } else { // it should never happen log.warn("[{}] lookup failed with error {}, {}", remoteAddress, topicName, ex.getMessage(), ex); - writeAndFlushVoidPromise(newLookupErrorResponse(ServerError.ServiceNotReady, + writeAndFlush(newLookupErrorResponse(ServerError.ServiceNotReady, ex.getMessage(), requestId)); } lookupSemaphore.release(); @@ -516,14 +517,14 @@ protected void handleLookup(CommandLookupTopic lookup) { } else { final String msg = "Client is not authorized to Lookup"; log.warn("[{}] {} with role {} on topic {}", remoteAddress, msg, getPrincipal(), topicName); - writeAndFlushVoidPromise(newLookupErrorResponse(ServerError.AuthorizationError, msg, requestId)); + writeAndFlush(newLookupErrorResponse(ServerError.AuthorizationError, msg, requestId)); lookupSemaphore.release(); } return null; }).exceptionally(ex -> { logAuthException(remoteAddress, "lookup", getPrincipal(), Optional.of(topicName), ex); final String msg = "Exception occurred while trying to authorize lookup"; - writeAndFlushVoidPromise(newLookupErrorResponse(ServerError.AuthorizationError, msg, requestId)); + writeAndFlush(newLookupErrorResponse(ServerError.AuthorizationError, msg, requestId)); lookupSemaphore.release(); return null; }); @@ -531,13 +532,13 @@ protected void handleLookup(CommandLookupTopic lookup) { if (log.isDebugEnabled()) { log.debug("[{}] Failed lookup due to too many lookup-requests {}", remoteAddress, topicName); } - writeAndFlushVoidPromise(newLookupErrorResponse(ServerError.TooManyRequests, + writeAndFlush(newLookupErrorResponse(ServerError.TooManyRequests, "Failed due to too many pending lookup requests", requestId)); } } - private void writeAndFlushVoidPromise(ByteBuf cmd) { - ctx.writeAndFlush(cmd, ctx.voidPromise()); + private void writeAndFlush(ByteBuf cmd) { + NettyChannelUtil.writeAndFlushWithVoidPromise(ctx, cmd); } @Override @@ -561,7 +562,7 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa partitionMetadata.getTopic(), remoteAddress, requestId, this.service.getPulsar().getState().toString()); } - writeAndFlushVoidPromise(Commands.newPartitionMetadataResponse(ServerError.ServiceNotReady, + writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.ServiceNotReady, "Failed due to pulsar service is not ready", requestId)); return; } @@ -611,14 +612,15 @@ protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata pa } else { final String msg = "Client is not authorized to Get Partition Metadata"; log.warn("[{}] {} with role {} on topic {}", remoteAddress, msg, getPrincipal(), topicName); - writeAndFlushVoidPromise(Commands.newPartitionMetadataResponse(ServerError.AuthorizationError, msg, requestId)); + writeAndFlush( + Commands.newPartitionMetadataResponse(ServerError.AuthorizationError, msg, requestId)); lookupSemaphore.release(); } return null; }).exceptionally(ex -> { logAuthException(remoteAddress, "partition-metadata", getPrincipal(), Optional.of(topicName), ex); final String msg = "Exception occurred while trying to authorize get Partition Metadata"; - writeAndFlushVoidPromise(Commands.newPartitionMetadataResponse(ServerError.AuthorizationError, msg, + writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.AuthorizationError, msg, requestId)); lookupSemaphore.release(); return null; @@ -660,7 +662,7 @@ protected void handleConsumerStats(CommandConsumerStats commandConsumerStats) { msg = createConsumerStatsResponse(consumer, requestId); } - writeAndFlushVoidPromise(msg); + writeAndFlush(msg); } ByteBuf createConsumerStatsResponse(Consumer consumer, long requestId) { @@ -690,7 +692,7 @@ ByteBuf createConsumerStatsResponse(Consumer consumer, long requestId) { // complete the connect and sent newConnected command private void completeConnect(int clientProtoVersion, String clientVersion, boolean supportsTopicWatchers) { - writeAndFlushVoidPromise(Commands.newConnected(clientProtoVersion, maxMessageSize, supportsTopicWatchers)); + writeAndFlush(Commands.newConnected(clientProtoVersion, maxMessageSize, supportsTopicWatchers)); state = State.Connected; service.getPulsarStats().recordConnectionCreateSuccess(); if (log.isDebugEnabled()) { @@ -766,7 +768,7 @@ private State doAuthentication(AuthData clientData, } // auth not complete, continue auth with client side. - writeAndFlushVoidPromise(Commands.newAuthChallenge(authMethod, brokerData, clientProtocolVersion)); + writeAndFlush(Commands.newAuthChallenge(authMethod, brokerData, clientProtocolVersion)); if (log.isDebugEnabled()) { log.debug("[{}] Authentication in progress client by method {}.", remoteAddress, authMethod); @@ -816,7 +818,7 @@ public void refreshAuthenticationCredentials() { try { AuthData brokerData = authState.refreshAuthentication(); - writeAndFlushVoidPromise(Commands.newAuthChallenge(authMethod, brokerData, + writeAndFlush(Commands.newAuthChallenge(authMethod, brokerData, getRemoteEndpointProtocolVersion())); if (log.isDebugEnabled()) { log.debug("[{}] Sent auth challenge to client to refresh credentials with method: {}.", @@ -852,7 +854,12 @@ protected void handleConnect(CommandConnect connect) { log.debug("Failed CONNECT from {} due to pulsar service is not ready: {} state", remoteAddress, this.service.getPulsar().getState().toString()); } - writeAndFlushVoidPromise(Commands.newError(-1, ServerError.ServiceNotReady, "Failed due to pulsar service is not ready")); + writeAndFlush( + Commands.newError( + -1, + ServerError.ServiceNotReady, + "Failed due to pulsar service is not ready") + ); close(); return; } @@ -963,7 +970,7 @@ protected void handleConnect(CommandConnect connect) { service.getPulsarStats().recordConnectionCreateFail(); logAuthException(remoteAddress, "connect", getPrincipal(), Optional.empty(), e); String msg = "Unable to authenticate"; - writeAndFlushVoidPromise(Commands.newError(-1, ServerError.AuthenticationError, msg)); + writeAndFlush(Commands.newError(-1, ServerError.AuthenticationError, msg)); close(); } } @@ -987,13 +994,13 @@ protected void handleAuthResponse(CommandAuthResponse authResponse) { } catch (AuthenticationException e) { service.getPulsarStats().recordConnectionCreateFail(); log.warn("[{}] Authentication failed: {} ", remoteAddress, e.getMessage()); - writeAndFlushVoidPromise(Commands.newError(-1, ServerError.AuthenticationError, e.getMessage())); + writeAndFlush(Commands.newError(-1, ServerError.AuthenticationError, e.getMessage())); close(); } catch (Exception e) { service.getPulsarStats().recordConnectionCreateFail(); String msg = "Unable to handleAuthResponse"; log.warn("[{}] {} ", remoteAddress, msg, e); - writeAndFlushVoidPromise(Commands.newError(-1, ServerError.UnknownError, msg)); + writeAndFlush(Commands.newError(-1, ServerError.UnknownError, msg)); close(); } } @@ -1213,7 +1220,7 @@ protected void handleSubscribe(final CommandSubscribe subscribe) { String msg = "Client is not authorized to subscribe"; log.warn("[{}] {} with role {}", remoteAddress, msg, getPrincipal()); consumers.remove(consumerId, consumerFuture); - writeAndFlushVoidPromise(Commands.newError(requestId, ServerError.AuthorizationError, msg)); + writeAndFlush(Commands.newError(requestId, ServerError.AuthorizationError, msg)); } return null; }).exceptionally(ex -> { @@ -1290,7 +1297,7 @@ protected void handleProducer(final CommandProducer cmdProducer) { if (!isAuthorized) { String msg = "Client is not authorized to Produce"; log.warn("[{}] {} with role {}", remoteAddress, msg, getPrincipal()); - writeAndFlushVoidPromise(Commands.newError(requestId, ServerError.AuthorizationError, msg)); + writeAndFlush(Commands.newError(requestId, ServerError.AuthorizationError, msg)); return null; } @@ -1660,7 +1667,7 @@ protected void handleAck(CommandAck ack) { Consumer consumer = consumerFuture.getNow(null); consumer.messageAcked(ack).thenRun(() -> { if (hasRequestId) { - writeAndFlushVoidPromise(Commands.newAckResponse( + writeAndFlush(Commands.newAckResponse( requestId, null, null, consumerId)); } if (brokerInterceptor != null) { @@ -1668,7 +1675,7 @@ protected void handleAck(CommandAck ack) { } }).exceptionally(e -> { if (hasRequestId) { - writeAndFlushVoidPromise(Commands.newAckResponse(requestId, + writeAndFlush(Commands.newAckResponse(requestId, BrokerServiceException.getClientErrorCode(e), e.getMessage(), consumerId)); } @@ -1827,7 +1834,7 @@ protected void handleCloseProducer(CommandCloseProducer closeProducer) { CompletableFuture producerFuture = producers.get(producerId); if (producerFuture == null) { log.info("[{}] Producer {} was not registered on the connection", remoteAddress, producerId); - writeAndFlushVoidPromise(Commands.newSuccess(requestId)); + writeAndFlush(Commands.newSuccess(requestId)); return; } @@ -1876,7 +1883,7 @@ protected void handleCloseConsumer(CommandCloseConsumer closeConsumer) { CompletableFuture consumerFuture = consumers.get(consumerId); if (consumerFuture == null) { log.info("[{}] Consumer was not registered on the connection: {}", consumerId, remoteAddress); - writeAndFlushVoidPromise(Commands.newSuccess(requestId)); + writeAndFlush(Commands.newSuccess(requestId)); return; } @@ -1943,7 +1950,7 @@ protected void handleGetLastMessageId(CommandGetLastMessageId getLastMessageId) consumer.getSubscription().getName()); } else { - writeAndFlushVoidPromise(Commands.newError(getLastMessageId.getRequestId(), + writeAndFlush(Commands.newError(getLastMessageId.getRequestId(), ServerError.MetadataError, "Consumer not found")); } } @@ -1996,7 +2003,7 @@ public void readEntryFailed(ManagedLedgerException exception, Object ctx) { handleLastMessageIdFromCompactedLedger(persistentTopic, requestId, partitionIndex, markDeletePosition); } else { - writeAndFlushVoidPromise(Commands.newError( + writeAndFlush(Commands.newError( requestId, ServerError.MetadataError, "Failed to get batch size for entry " + e.getMessage())); } @@ -2008,7 +2015,7 @@ public void readEntryFailed(ManagedLedgerException exception, Object ctx) { topic.getName(), subscriptionName, lastPosition, partitionIndex); } - writeAndFlushVoidPromise(Commands.newGetLastMessageIdResponse(requestId, lastPosition.getLedgerId(), + writeAndFlush(Commands.newGetLastMessageIdResponse(requestId, lastPosition.getLedgerId(), lastPosition.getEntryId(), partitionIndex, largestBatchIndex, markDeletePosition != null ? markDeletePosition.getLedgerId() : -1, markDeletePosition != null ? markDeletePosition.getEntryId() : -1)); @@ -2028,12 +2035,12 @@ private void handleLastMessageIdFromCompactedLedger(PersistentTopic persistentTo try { largestBatchIndex = calculateTheLastBatchIndexInBatch(metadata, payload); } catch (IOException ioEx){ - writeAndFlushVoidPromise(Commands.newError(requestId, ServerError.MetadataError, + writeAndFlush(Commands.newError(requestId, ServerError.MetadataError, "Failed to deserialize batched message from the last entry of the compacted Ledger: " + ioEx.getMessage())); return; } - writeAndFlushVoidPromise(Commands.newGetLastMessageIdResponse(requestId, + writeAndFlush(Commands.newGetLastMessageIdResponse(requestId, entry.getLedgerId(), entry.getEntryId(), partitionIndex, largestBatchIndex, markDeletePosition != null ? markDeletePosition.getLedgerId() : -1, markDeletePosition != null ? markDeletePosition.getEntryId() : -1)); @@ -2041,13 +2048,13 @@ private void handleLastMessageIdFromCompactedLedger(PersistentTopic persistentTo } else { // in this case, the ledgers been removed except the current ledger // and current ledger without any data - writeAndFlushVoidPromise(Commands.newGetLastMessageIdResponse(requestId, + writeAndFlush(Commands.newGetLastMessageIdResponse(requestId, -1, -1, partitionIndex, -1, markDeletePosition != null ? markDeletePosition.getLedgerId() : -1, markDeletePosition != null ? markDeletePosition.getEntryId() : -1)); } }).exceptionally(ex -> { - writeAndFlushVoidPromise(Commands.newError( + writeAndFlush(Commands.newError( requestId, ServerError.MetadataError, "Failed to read last entry of the compacted Ledger " + ex.getCause().getMessage())); @@ -2403,12 +2410,12 @@ protected void handleAddPartitionToTxn(CommandAddPartitionToTxn command) { if (log.isDebugEnabled()) { log.debug("Send response success for add published partition to txn request {}", requestId); } - writeAndFlushVoidPromise(Commands.newAddPartitionToTxnResponse(requestId, + writeAndFlush(Commands.newAddPartitionToTxnResponse(requestId, txnID.getLeastSigBits(), txnID.getMostSigBits())); } else { ex = handleTxnException(ex, BaseCommand.Type.ADD_PARTITION_TO_TXN.name(), requestId); - writeAndFlushVoidPromise(Commands.newAddPartitionToTxnResponse(requestId, + writeAndFlush(Commands.newAddPartitionToTxnResponse(requestId, txnID.getLeastSigBits(), txnID.getMostSigBits(), BrokerServiceException.getClientErrorCode(ex), @@ -2470,13 +2477,13 @@ protected void handleEndTxnOnPartition(CommandEndTxnOnPartition command) { if (throwable != null) { log.error("handleEndTxnOnPartition fail!, topic {}, txnId: [{}], " + "txnAction: [{}]", topic, txnID, TxnAction.valueOf(txnAction), throwable); - writeAndFlushVoidPromise(Commands.newEndTxnOnPartitionResponse( + writeAndFlush(Commands.newEndTxnOnPartitionResponse( requestId, BrokerServiceException.getClientErrorCode(throwable), throwable.getMessage(), txnID.getLeastSigBits(), txnID.getMostSigBits())); return; } - writeAndFlushVoidPromise(Commands.newEndTxnOnPartitionResponse(requestId, + writeAndFlush(Commands.newEndTxnOnPartitionResponse(requestId, txnID.getLeastSigBits(), txnID.getMostSigBits())); }); @@ -2488,7 +2495,7 @@ protected void handleEndTxnOnPartition(CommandEndTxnOnPartition command) { log.error("handleEndTxnOnPartition fail ! The topic {} does not exist in broker, " + "txnId: [{}], txnAction: [{}]", topic, txnID, TxnAction.valueOf(txnAction)); - writeAndFlushVoidPromise(Commands.newEndTxnOnPartitionResponse(requestId, + writeAndFlush(Commands.newEndTxnOnPartitionResponse(requestId, ServerError.ServiceNotReady, "The topic " + topic + " does not exist in broker.", txnID.getLeastSigBits(), txnID.getMostSigBits())); @@ -2496,14 +2503,14 @@ protected void handleEndTxnOnPartition(CommandEndTxnOnPartition command) { log.warn("handleEndTxnOnPartition fail ! The topic {} has not been created, " + "txnId: [{}], txnAction: [{}]", topic, txnID, TxnAction.valueOf(txnAction)); - writeAndFlushVoidPromise(Commands.newEndTxnOnPartitionResponse(requestId, + writeAndFlush(Commands.newEndTxnOnPartitionResponse(requestId, txnID.getLeastSigBits(), txnID.getMostSigBits())); } }).exceptionally(e -> { log.error("handleEndTxnOnPartition fail ! topic {}, " + "txnId: [{}], txnAction: [{}]", topic, txnID, TxnAction.valueOf(txnAction), e.getCause()); - writeAndFlushVoidPromise(Commands.newEndTxnOnPartitionResponse( + writeAndFlush(Commands.newEndTxnOnPartitionResponse( requestId, ServerError.ServiceNotReady, e.getMessage(), txnID.getLeastSigBits(), txnID.getMostSigBits())); return null; @@ -2513,7 +2520,7 @@ protected void handleEndTxnOnPartition(CommandEndTxnOnPartition command) { log.error("handleEndTxnOnPartition fail ! topic {}, " + "txnId: [{}], txnAction: [{}]", topic, txnID, TxnAction.valueOf(txnAction), e.getCause()); - writeAndFlushVoidPromise(Commands.newEndTxnOnPartitionResponse( + writeAndFlush(Commands.newEndTxnOnPartitionResponse( requestId, ServerError.ServiceNotReady, e.getMessage(), txnID.getLeastSigBits(), txnID.getMostSigBits())); return null; @@ -2545,7 +2552,8 @@ protected void handleEndTxnOnSubscription(CommandEndTxnOnSubscription command) { log.warn("handleEndTxnOnSubscription fail! " + "topic {} subscription {} does not exist. txnId: [{}], txnAction: [{}]", optionalTopic.get().getName(), subName, txnID, TxnAction.valueOf(txnAction)); - writeAndFlushVoidPromise(Commands.newEndTxnOnSubscriptionResponse(requestId, txnidLeastBits, txnidMostBits)); + writeAndFlush( + Commands.newEndTxnOnSubscriptionResponse(requestId, txnidLeastBits, txnidMostBits)); return; } @@ -2556,13 +2564,14 @@ protected void handleEndTxnOnSubscription(CommandEndTxnOnSubscription command) { log.error("handleEndTxnOnSubscription fail ! topic: {}, subscription: {}" + "txnId: [{}], txnAction: [{}]", topic, subName, txnID, TxnAction.valueOf(txnAction), e.getCause()); - writeAndFlushVoidPromise(Commands.newEndTxnOnSubscriptionResponse( + writeAndFlush(Commands.newEndTxnOnSubscriptionResponse( requestId, txnidLeastBits, txnidMostBits, BrokerServiceException.getClientErrorCode(e), "Handle end txn on subscription failed: " + e.getMessage())); return; } - writeAndFlushVoidPromise(Commands.newEndTxnOnSubscriptionResponse(requestId, txnidLeastBits, txnidMostBits)); + writeAndFlush( + Commands.newEndTxnOnSubscriptionResponse(requestId, txnidLeastBits, txnidMostBits)); }); } else { getBrokerService().getManagedLedgerFactory() @@ -2572,7 +2581,7 @@ protected void handleEndTxnOnSubscription(CommandEndTxnOnSubscription command) { log.error("handleEndTxnOnSubscription fail! The topic {} does not exist in broker, " + "subscription: {}, txnId: [{}], txnAction: [{}]", topic, subName, txnID, TxnAction.valueOf(txnAction)); - writeAndFlushVoidPromise(Commands.newEndTxnOnSubscriptionResponse( + writeAndFlush(Commands.newEndTxnOnSubscriptionResponse( requestId, txnID.getLeastSigBits(), txnID.getMostSigBits(), ServerError.ServiceNotReady, "The topic " + topic + " does not exist in broker.")); @@ -2580,14 +2589,14 @@ protected void handleEndTxnOnSubscription(CommandEndTxnOnSubscription command) { log.warn("handleEndTxnOnSubscription fail ! The topic {} has not been created, " + "subscription: {} txnId: [{}], txnAction: [{}]", topic, subName, txnID, TxnAction.valueOf(txnAction)); - writeAndFlushVoidPromise(Commands.newEndTxnOnSubscriptionResponse(requestId, + writeAndFlush(Commands.newEndTxnOnSubscriptionResponse(requestId, txnID.getLeastSigBits(), txnID.getMostSigBits())); } }).exceptionally(e -> { log.error("handleEndTxnOnSubscription fail ! topic {}, subscription: {}" + "txnId: [{}], txnAction: [{}]", topic, subName, txnID, TxnAction.valueOf(txnAction), e.getCause()); - writeAndFlushVoidPromise(Commands.newEndTxnOnSubscriptionResponse( + writeAndFlush(Commands.newEndTxnOnSubscriptionResponse( requestId, txnID.getLeastSigBits(), txnID.getMostSigBits(), ServerError.ServiceNotReady, e.getMessage())); return null; @@ -2597,7 +2606,7 @@ protected void handleEndTxnOnSubscription(CommandEndTxnOnSubscription command) { log.error("handleEndTxnOnSubscription fail ! topic: {}, subscription: {}" + "txnId: [{}], txnAction: [{}]", topic, subName, txnID, TxnAction.valueOf(txnAction), e.getCause()); - writeAndFlushVoidPromise(Commands.newEndTxnOnSubscriptionResponse( + writeAndFlush(Commands.newEndTxnOnSubscriptionResponse( requestId, txnidLeastBits, txnidMostBits, ServerError.ServiceNotReady, "Handle end txn on subscription failed: " + e.getMessage())); @@ -2652,11 +2661,11 @@ protected void handleAddSubscriptionToTxn(CommandAddSubscriptionToTxn command) { log.debug("Send response success for add published partition to txn request {}", requestId); } - writeAndFlushVoidPromise(Commands.newAddSubscriptionToTxnResponse(requestId, + writeAndFlush(Commands.newAddSubscriptionToTxnResponse(requestId, txnID.getLeastSigBits(), txnID.getMostSigBits())); } else { ex = handleTxnException(ex, BaseCommand.Type.ADD_SUBSCRIPTION_TO_TXN.name(), requestId); - writeAndFlushVoidPromise( + writeAndFlush( Commands.newAddSubscriptionToTxnResponse(requestId, txnID.getLeastSigBits(), txnID.getMostSigBits(), BrokerServiceException.getClientErrorCode(ex), ex.getMessage())); @@ -2743,7 +2752,7 @@ public void closeProducer(Producer producer) { // removes producer-connection from map and send close command to producer safelyRemoveProducer(producer); if (getRemoteEndpointProtocolVersion() >= v5.getValue()) { - writeAndFlushVoidPromise(Commands.newCloseProducer(producer.getProducerId(), -1L)); + writeAndFlush(Commands.newCloseProducer(producer.getProducerId(), -1L)); } else { close(); } @@ -2755,7 +2764,7 @@ public void closeConsumer(Consumer consumer) { // removes consumer-connection from map and send close command to consumer safelyRemoveConsumer(consumer); if (getRemoteEndpointProtocolVersion() >= v5.getValue()) { - writeAndFlushVoidPromise(Commands.newCloseConsumer(consumer.consumerId(), -1L)); + writeAndFlush(Commands.newCloseConsumer(consumer.consumerId(), -1L)); } else { close(); } @@ -2998,13 +3007,13 @@ private TopicName validateTopicName(String topic, long requestId, Object request } if (requestCommand instanceof CommandLookupTopic) { - writeAndFlushVoidPromise(Commands.newLookupErrorResponse(ServerError.InvalidTopicName, + writeAndFlush(Commands.newLookupErrorResponse(ServerError.InvalidTopicName, "Invalid topic name: " + t.getMessage(), requestId)); } else if (requestCommand instanceof CommandPartitionedTopicMetadata) { - writeAndFlushVoidPromise(Commands.newPartitionMetadataResponse(ServerError.InvalidTopicName, + writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.InvalidTopicName, "Invalid topic name: " + t.getMessage(), requestId)); } else { - writeAndFlushVoidPromise(Commands.newError(requestId, ServerError.InvalidTopicName, + writeAndFlush(Commands.newError(requestId, ServerError.InvalidTopicName, "Invalid topic name: " + t.getMessage())); } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java index e122e557c491d..496652fed0b6b 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java @@ -85,6 +85,7 @@ import org.apache.pulsar.common.api.proto.CommandWatchTopicUpdate; import org.apache.pulsar.common.api.proto.ServerError; import org.apache.pulsar.common.intercept.InterceptException; +import org.apache.pulsar.common.util.netty.NettyChannelUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -132,7 +133,8 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception interceptCommand(cmd); handlePartitionMetadataRequest(cmd.getPartitionMetadata()); } catch (InterceptException e) { - writeAndFlushVoidPromise(ctx, Commands.newPartitionMetadataResponse(getServerError(e.getErrorCode()), + writeAndFlush(ctx, + Commands.newPartitionMetadataResponse(getServerError(e.getErrorCode()), e.getMessage(), cmd.getPartitionMetadata().getRequestId())); } break; @@ -206,7 +208,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception interceptCommand(cmd); handleProducer(cmd.getProducer()); } catch (InterceptException e) { - writeAndFlushVoidPromise(ctx, Commands.newError(cmd.getProducer().getRequestId(), + writeAndFlush(ctx, Commands.newError(cmd.getProducer().getRequestId(), getServerError(e.getErrorCode()), e.getMessage())); } break; @@ -219,7 +221,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception ByteBuf headersAndPayload = buffer.markReaderIndex(); handleSend(cmd.getSend(), headersAndPayload); } catch (InterceptException e) { - writeAndFlushVoidPromise(ctx, Commands.newSendError(cmd.getSend().getProducerId(), + writeAndFlush(ctx, Commands.newSendError(cmd.getSend().getProducerId(), cmd.getSend().getSequenceId(), getServerError(e.getErrorCode()), e.getMessage())); } break; @@ -240,7 +242,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception interceptCommand(cmd); handleSubscribe(cmd.getSubscribe()); } catch (InterceptException e) { - writeAndFlushVoidPromise(ctx, Commands.newError(cmd.getSubscribe().getRequestId(), + writeAndFlush(ctx, Commands.newError(cmd.getSubscribe().getRequestId(), getServerError(e.getErrorCode()), e.getMessage())); } break; @@ -267,8 +269,13 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception interceptCommand(cmd); handleSeek(cmd.getSeek()); } catch (InterceptException e) { - writeAndFlushVoidPromise(ctx, Commands.newError(cmd.getSeek().getRequestId(), getServerError(e.getErrorCode()), - e.getMessage())); + writeAndFlush(ctx, + Commands.newError( + cmd.getSeek().getRequestId(), + getServerError(e.getErrorCode()), + e.getMessage() + ) + ); } break; @@ -327,7 +334,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception interceptCommand(cmd); handleGetTopicsOfNamespace(cmd.getGetTopicsOfNamespace()); } catch (InterceptException e) { - writeAndFlushVoidPromise(ctx, Commands.newError(cmd.getGetTopicsOfNamespace().getRequestId(), + writeAndFlush(ctx, Commands.newError(cmd.getGetTopicsOfNamespace().getRequestId(), getServerError(e.getErrorCode()), e.getMessage())); } break; @@ -343,7 +350,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception interceptCommand(cmd); handleGetSchema(cmd.getGetSchema()); } catch (InterceptException e) { - writeAndFlushVoidPromise(ctx, Commands.newGetSchemaResponseError(cmd.getGetSchema().getRequestId(), + writeAndFlush(ctx, Commands.newGetSchemaResponseError(cmd.getGetSchema().getRequestId(), getServerError(e.getErrorCode()), e.getMessage())); } break; @@ -359,7 +366,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception interceptCommand(cmd); handleGetOrCreateSchema(cmd.getGetOrCreateSchema()); } catch (InterceptException e) { - writeAndFlushVoidPromise(ctx, Commands.newGetOrCreateSchemaResponseError( + writeAndFlush(ctx, Commands.newGetOrCreateSchemaResponseError( cmd.getGetOrCreateSchema().getRequestId(), getServerError(e.getErrorCode()), e.getMessage())); } @@ -733,7 +740,7 @@ protected void handleCommandWatchTopicListClose(CommandWatchTopicListClose comma private static final Logger log = LoggerFactory.getLogger(PulsarDecoder.class); - private void writeAndFlushVoidPromise(ChannelOutboundInvoker ctx, ByteBuf cmd) { - ctx.writeAndFlush(cmd, ctx.voidPromise()); + private void writeAndFlush(ChannelOutboundInvoker ctx, ByteBuf cmd) { + NettyChannelUtil.writeAndFlushWithVoidPromise(ctx, cmd); } } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/NettyChannelUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/NettyChannelUtil.java new file mode 100644 index 0000000000000..2f7061ef4ca54 --- /dev/null +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/NettyChannelUtil.java @@ -0,0 +1,38 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.common.util.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelOutboundInvoker; + +/** + * Contains utility methods for working with Netty Channels. + */ +public class NettyChannelUtil { + + public static void writeAndFlushWithVoidPromise(ChannelOutboundInvoker ctx, ByteBuf cmd) { + ctx.writeAndFlush(cmd, ctx.voidPromise()); + } + + public static void writeAndFlushWithClosePromise(ChannelOutboundInvoker ctx, ByteBuf cmd) { + ctx.writeAndFlush(cmd).addListener(ChannelFutureListener.CLOSE); + } + +} diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/NettyChannelUtilTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/NettyChannelUtilTest.java new file mode 100644 index 0000000000000..53a5f91de1d52 --- /dev/null +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/netty/NettyChannelUtilTest.java @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.common.util.netty; + +import static org.mockito.ArgumentMatchers.same; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelOutboundInvoker; +import io.netty.channel.ChannelPromise; +import io.netty.channel.VoidChannelPromise; +import java.nio.charset.StandardCharsets; +import org.testng.annotations.Test; + +public class NettyChannelUtilTest { + + @Test + public void testWriteAndFlushWithVoidPromise() { + final ChannelOutboundInvoker ctx = mock(ChannelOutboundInvoker.class); + final VoidChannelPromise voidChannelPromise = mock(VoidChannelPromise.class); + when(ctx.voidPromise()).thenReturn(voidChannelPromise); + final byte[] data = "test".getBytes(StandardCharsets.UTF_8); + final ByteBuf byteBuf = Unpooled.wrappedBuffer(data, 0, data.length); + try { + NettyChannelUtil.writeAndFlushWithVoidPromise(ctx, byteBuf); + verify(ctx).writeAndFlush(same(byteBuf), same(voidChannelPromise)); + verify(ctx).voidPromise(); + } finally { + byteBuf.release(); + } + } + + @Test + public void testWriteAndFlushWithClosePromise() { + final ChannelOutboundInvoker ctx = mock(ChannelOutboundInvoker.class); + final ChannelPromise promise = mock(ChannelPromise.class); + + final byte[] data = "test".getBytes(StandardCharsets.UTF_8); + final ByteBuf byteBuf = Unpooled.wrappedBuffer(data, 0, data.length); + when(ctx.writeAndFlush(same(byteBuf))).thenReturn(promise); + try { + NettyChannelUtil.writeAndFlushWithClosePromise(ctx, byteBuf); + verify(ctx).writeAndFlush(same(byteBuf)); + verify(promise).addListener(same(ChannelFutureListener.CLOSE)); + } finally { + byteBuf.release(); + } + } +} \ No newline at end of file diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java index d51f17b4731fa..4b5fef3a994bd 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java @@ -26,7 +26,6 @@ import io.netty.buffer.Unpooled; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; -import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInitializer; @@ -65,6 +64,7 @@ import org.apache.pulsar.common.util.SecurityUtility; import org.apache.pulsar.common.util.SslContextAutoRefreshBuilder; import org.apache.pulsar.common.util.keystoretls.NettySSLContextAutoRefreshBuilder; +import org.apache.pulsar.common.util.netty.NettyChannelUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -238,8 +238,8 @@ private static String parseHost(String brokerPortAndHost) { private void writeHAProxyMessage() { if (proxyConnection.hasHAProxyMessage()) { - outboundChannel.writeAndFlush(encodeProxyProtocolMessage(proxyConnection.getHAProxyMessage())) - .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); + final ByteBuf msg = encodeProxyProtocolMessage(proxyConnection.getHAProxyMessage()); + writeAndFlush(msg); } else { if (inboundChannel.remoteAddress() instanceof InetSocketAddress && inboundChannel.localAddress() instanceof InetSocketAddress) { @@ -252,8 +252,8 @@ private void writeHAProxyMessage() { HAProxyMessage msg = new HAProxyMessage(HAProxyProtocolVersion.V1, HAProxyCommand.PROXY, HAProxyProxiedProtocol.TCP4, sourceAddress, destinationAddress, sourcePort, destinationPort); - outboundChannel.writeAndFlush(encodeProxyProtocolMessage(msg)) - .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); + final ByteBuf encodedMsg = encodeProxyProtocolMessage(msg); + writeAndFlush(encodedMsg); msg.release(); } } @@ -323,11 +323,11 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception { // Send the Connect command to broker authenticationDataProvider = authentication.getAuthData(remoteHostName); AuthData authData = authenticationDataProvider.authenticate(AuthData.INIT_AUTH_DATA); - ByteBuf command; - command = Commands.newConnect(authentication.getAuthMethodName(), authData, protocolVersion, "Pulsar proxy", - null /* target broker */, originalPrincipal, clientAuthData, clientAuthMethod); - outboundChannel.writeAndFlush(command) - .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); + ByteBuf command = Commands.newConnect( + authentication.getAuthMethodName(), authData, protocolVersion, + "Pulsar proxy", null /* target broker */, + originalPrincipal, clientAuthData, clientAuthMethod); + writeAndFlush(command); isTlsOutboundChannel = ProxyConnection.isTlsChannel(inboundChannel); } @@ -358,8 +358,7 @@ public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exce if (msg instanceof ByteBuf) { ProxyService.BYTES_COUNTER.inc(((ByteBuf) msg).readableBytes()); } - inboundChannel.writeAndFlush(msg) - .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); + inboundChannel.writeAndFlush(msg, inboundChannel.voidPromise()); if (service.proxyZeroCopyModeEnabled && service.proxyLogLevel == 0) { if (!isTlsOutboundChannel && !DirectProxyHandler.this.proxyConnection.isTlsInboundChannel) { @@ -412,8 +411,7 @@ protected void handleAuthChallenge(CommandAuthChallenge authChallenge) { log.debug("{} Mutual auth {}", ctx.channel(), authentication.getAuthMethodName()); } - outboundChannel.writeAndFlush(request) - .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); + writeAndFlush(request); } catch (Exception e) { log.error("Error mutual verify", e); } @@ -495,5 +493,9 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { } } + private void writeAndFlush(ByteBuf cmd) { + NettyChannelUtil.writeAndFlushWithVoidPromise(outboundChannel, cmd); + } + private static final Logger log = LoggerFactory.getLogger(DirectProxyHandler.class); } diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java index 662afb18d1858..6ec597ec1cfc3 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java @@ -19,7 +19,7 @@ package org.apache.pulsar.proxy.server; import io.netty.buffer.ByteBuf; -import io.netty.channel.ChannelOutboundInvoker; +import io.netty.channel.ChannelHandlerContext; import io.prometheus.client.Counter; import java.net.InetSocketAddress; import java.net.SocketAddress; @@ -39,6 +39,7 @@ import org.apache.pulsar.common.protocol.Commands; import org.apache.pulsar.common.protocol.schema.BytesSchemaVersion; import org.apache.pulsar.common.protocol.schema.SchemaVersion; +import org.apache.pulsar.common.util.netty.NettyChannelUtil; import org.apache.pulsar.policies.data.loadbalancer.ServiceLookupData; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -114,7 +115,7 @@ public void handleLookup(CommandLookupTopic lookup) { log.debug("Lookup Request ID {} from {} rejected - {}.", clientRequestId, clientAddress, throttlingErrorMessage); } - writeAndFlushVoidPromise(Commands.newLookupErrorResponse(ServerError.ServiceNotReady, + writeAndFlush(Commands.newLookupErrorResponse(ServerError.ServiceNotReady, throttlingErrorMessage, clientRequestId)); } @@ -123,7 +124,7 @@ public void handleLookup(CommandLookupTopic lookup) { private void performLookup(long clientRequestId, String topic, String brokerServiceUrl, boolean authoritative, int numberOfRetries) { if (numberOfRetries == 0) { - writeAndFlushVoidPromise(Commands.newLookupErrorResponse(ServerError.ServiceNotReady, + writeAndFlush(Commands.newLookupErrorResponse(ServerError.ServiceNotReady, "Reached max number of redirections", clientRequestId)); return; } @@ -132,7 +133,7 @@ private void performLookup(long clientRequestId, String topic, String brokerServ try { brokerURI = new URI(brokerServiceUrl); } catch (URISyntaxException e) { - writeAndFlushVoidPromise( + writeAndFlush( Commands.newLookupErrorResponse(ServerError.MetadataError, e.getMessage(), clientRequestId)); return; } @@ -151,7 +152,7 @@ private void performLookup(long clientRequestId, String topic, String brokerServ clientCnx.newLookup(command, requestId).whenComplete((r, t) -> { if (t != null) { log.warn("[{}] Failed to lookup topic {}: {}", clientAddress, topic, t.getMessage()); - writeAndFlushVoidPromise( + writeAndFlush( Commands.newLookupErrorResponse(getServerError(t), t.getMessage(), clientRequestId)); } else { String brokerUrl = connectWithTLS ? r.brokerUrlTls : r.brokerUrl; @@ -171,7 +172,7 @@ private void performLookup(long clientRequestId, String topic, String brokerServ + " with clientReq Id '{}' and lookup-broker {}", addr, topic, clientRequestId, brokerUrl); } - writeAndFlushVoidPromise(Commands.newLookupResponse(brokerUrl, brokerUrl, true, + writeAndFlush(Commands.newLookupResponse(brokerUrl, brokerUrl, true, LookupType.Connect, clientRequestId, true /* this is coming from proxy */)); } } @@ -179,7 +180,7 @@ private void performLookup(long clientRequestId, String topic, String brokerServ }); }).exceptionally(ex -> { // Failed to connect to backend broker - writeAndFlushVoidPromise( + writeAndFlush( Commands.newLookupErrorResponse(getServerError(ex), ex.getMessage(), clientRequestId)); return null; }); @@ -203,7 +204,7 @@ public void handlePartitionMetadataResponse(CommandPartitionedTopicMetadata part log.debug("PartitionMetaData Request ID {} from {} rejected - {}.", clientRequestId, clientAddress, throttlingErrorMessage); } - writeAndFlushVoidPromise(Commands.newPartitionMetadataResponse(ServerError.ServiceNotReady, + writeAndFlush(Commands.newPartitionMetadataResponse(ServerError.ServiceNotReady, throttlingErrorMessage, clientRequestId)); } } @@ -240,17 +241,17 @@ private void handlePartitionMetadataResponse(CommandPartitionedTopicMetadata par if (t != null) { log.warn("[{}] failed to get Partitioned metadata : {}", topicName.toString(), t.getMessage(), t); - writeAndFlushVoidPromise(Commands.newLookupErrorResponse(getServerError(t), + writeAndFlush(Commands.newLookupErrorResponse(getServerError(t), t.getMessage(), clientRequestId)); } else { - writeAndFlushVoidPromise( + writeAndFlush( Commands.newPartitionMetadataResponse(r.partitions, clientRequestId)); } proxyConnection.getConnectionPool().releaseConnection(clientCnx); }); }).exceptionally(ex -> { // Failed to connect to backend broker - writeAndFlushVoidPromise(Commands.newPartitionMetadataResponse(getServerError(ex), + writeAndFlush(Commands.newPartitionMetadataResponse(getServerError(ex), ex.getMessage(), clientRequestId)); return null; }); @@ -276,7 +277,7 @@ public void handleGetTopicsOfNamespace(CommandGetTopicsOfNamespace commandGetTop log.debug("GetTopicsOfNamespace Request ID {} from {} rejected - {}.", requestId, clientAddress, throttlingErrorMessage); } - writeAndFlushVoidPromise(Commands.newError( + writeAndFlush(Commands.newError( requestId, ServerError.ServiceNotReady, throttlingErrorMessage )); } @@ -305,7 +306,7 @@ private void performGetTopicsOfNamespace(long clientRequestId, String topicsHash, CommandGetTopicsOfNamespace.Mode mode) { if (numberOfRetries == 0) { - writeAndFlushVoidPromise(Commands.newError(clientRequestId, ServerError.ServiceNotReady, + writeAndFlush(Commands.newError(clientRequestId, ServerError.ServiceNotReady, "Reached max number of redirections")); return; } @@ -330,10 +331,10 @@ private void performGetTopicsOfNamespace(long clientRequestId, if (t != null) { log.warn("[{}] Failed to get TopicsOfNamespace {}: {}", clientAddress, namespaceName, t.getMessage()); - writeAndFlushVoidPromise( + writeAndFlush( Commands.newError(clientRequestId, getServerError(t), t.getMessage())); } else { - writeAndFlushVoidPromise( + writeAndFlush( Commands.newGetTopicsOfNamespaceResponse(r.getTopics(), r.getTopicsHash(), r.isFiltered(), r.isChanged(), clientRequestId)); } @@ -342,7 +343,7 @@ private void performGetTopicsOfNamespace(long clientRequestId, proxyConnection.getConnectionPool().releaseConnection(clientCnx); }).exceptionally(ex -> { // Failed to connect to backend broker - writeAndFlushVoidPromise( + writeAndFlush( Commands.newError(clientRequestId, getServerError(ex), ex.getMessage())); return null; }); @@ -385,10 +386,10 @@ public void handleGetSchema(CommandGetSchema commandGetSchema) { clientCnx.sendGetRawSchema(command, requestId).whenComplete((r, t) -> { if (t != null) { log.warn("[{}] Failed to get schema {}: {}", clientAddress, topic, t); - writeAndFlushVoidPromise( + writeAndFlush( Commands.newError(clientRequestId, getServerError(t), t.getMessage())); } else { - writeAndFlushVoidPromise( + writeAndFlush( Commands.newGetSchemaResponse(clientRequestId, r)); } @@ -396,7 +397,7 @@ public void handleGetSchema(CommandGetSchema commandGetSchema) { }); }).exceptionally(ex -> { // Failed to connect to backend broker - writeAndFlushVoidPromise( + writeAndFlush( Commands.newError(clientRequestId, getServerError(ex), ex.getMessage())); return null; }); @@ -415,7 +416,7 @@ private String getBrokerServiceUrl(long clientRequestId) { availableBroker = discoveryProvider.nextBroker(); } catch (Exception e) { log.warn("[{}] Failed to get next active broker {}", clientAddress, e.getMessage(), e); - writeAndFlushVoidPromise(Commands.newError( + writeAndFlush(Commands.newError( clientRequestId, ServerError.ServiceNotReady, e.getMessage() )); return null; @@ -428,7 +429,7 @@ private InetSocketAddress getAddr(String brokerServiceUrl, long clientRequestId) try { brokerURI = new URI(brokerServiceUrl); } catch (URISyntaxException e) { - writeAndFlushVoidPromise( + writeAndFlush( Commands.newError(clientRequestId, ServerError.MetadataError, e.getMessage())); return null; } @@ -447,8 +448,9 @@ private ServerError getServerError(Throwable error) { return responseError; } - private void writeAndFlushVoidPromise(ByteBuf cmd) { - proxyConnection.ctx().writeAndFlush(cmd); + private void writeAndFlush(ByteBuf cmd) { + final ChannelHandlerContext ctx = proxyConnection.ctx(); + NettyChannelUtil.writeAndFlushWithVoidPromise(ctx, cmd); } private static final Logger log = LoggerFactory.getLogger(LookupProxyHandler.class); diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java index 56fbe94606981..fe4d29ac4ab6b 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java @@ -71,6 +71,7 @@ import org.apache.pulsar.common.api.proto.ServerError; import org.apache.pulsar.common.protocol.Commands; import org.apache.pulsar.common.protocol.PulsarHandler; +import org.apache.pulsar.common.util.netty.NettyChannelUtil; import org.apache.pulsar.policies.data.loadbalancer.ServiceLookupData; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -261,8 +262,8 @@ public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exce directProxyHandler.getInboundChannelRequestsRate().recordEvent(bytes); ProxyService.BYTES_COUNTER.inc(bytes); } - directProxyHandler.outboundChannel.writeAndFlush(msg) - .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); + directProxyHandler.outboundChannel + .writeAndFlush(msg, directProxyHandler.outboundChannel.voidPromise()); if (service.proxyZeroCopyModeEnabled && service.proxyLogLevel == 0) { if (!directProxyHandler.isTlsOutboundChannel && !isTlsInboundChannel) { @@ -345,10 +346,9 @@ private synchronized void completeConnect(AuthData clientData) throws PulsarClie state = State.Closing; LOG.warn("[{}] Target broker '{}' isn't available. authenticated with {} role {}.", remoteAddress, proxyToBrokerUrl, authMethod, clientAuthRole); - ctx() - .writeAndFlush( - Commands.newError(-1, ServerError.ServiceNotReady, "Target broker isn't available.")) - .addListener(ChannelFutureListener.CLOSE); + final ByteBuf msg = Commands.newError(-1, + ServerError.ServiceNotReady, "Target broker isn't available."); + writeAndFlushAndClose(msg); return; } @@ -369,11 +369,9 @@ private synchronized void completeConnect(AuthData clientData) throws PulsarClie LOG.error("[{}] Error validating target broker '{}'. authenticated with {} role {}.", remoteAddress, proxyToBrokerUrl, authMethod, clientAuthRole, throwable); } - ctx() - .writeAndFlush( - Commands.newError(-1, ServerError.ServiceNotReady, - "Target broker cannot be validated.")) - .addListener(ChannelFutureListener.CLOSE); + final ByteBuf msg = Commands.newError(-1, ServerError.ServiceNotReady, + "Target broker cannot be validated."); + writeAndFlushAndClose(msg); return null; }); } else { @@ -382,8 +380,8 @@ private synchronized void completeConnect(AuthData clientData) throws PulsarClie // partitions metadata lookups state = State.ProxyLookupRequests; lookupProxyHandler = new LookupProxyHandler(service, this); - ctx.writeAndFlush(Commands.newConnected(protocolVersionToAdvertise, false)) - .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); + final ByteBuf msg = Commands.newConnected(protocolVersionToAdvertise, false); + writeAndFlush(msg); } } @@ -394,9 +392,9 @@ private void handleBrokerConnected(DirectProxyHandler directProxyHandler, Comman state = State.ProxyConnectionToBroker; int maxMessageSize = connected.hasMaxMessageSize() ? connected.getMaxMessageSize() : Commands.INVALID_MAX_MESSAGE_SIZE; - ctx.writeAndFlush(Commands.newConnected(connected.getProtocolVersion(), maxMessageSize, - connected.hasFeatureFlags() && connected.getFeatureFlags().isSupportsTopicWatchers())) - .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); + final ByteBuf msg = Commands.newConnected(connected.getProtocolVersion(), maxMessageSize, + connected.hasFeatureFlags() && connected.getFeatureFlags().isSupportsTopicWatchers()); + writeAndFlush(msg); } else { LOG.warn("[{}] Channel is {}. ProxyConnection is in {}. " + "Closing connection to broker '{}'.", @@ -445,8 +443,8 @@ private void doAuthentication(AuthData clientData) } // auth not complete, continue auth with client side. - ctx.writeAndFlush(Commands.newAuthChallenge(authMethod, brokerData, protocolVersionToAdvertise)) - .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); + final ByteBuf msg = Commands.newAuthChallenge(authMethod, brokerData, protocolVersionToAdvertise); + writeAndFlush(msg); if (LOG.isDebugEnabled()) { LOG.debug("[{}] Authentication in progress client by method {}.", remoteAddress, authMethod); @@ -524,8 +522,8 @@ remoteAddress, protocolVersionToAdvertise, getRemoteEndpointProtocolVersion(), doAuthentication(clientData); } catch (Exception e) { LOG.warn("[{}] Unable to authenticate: ", remoteAddress, e); - ctx.writeAndFlush(Commands.newError(-1, ServerError.AuthenticationError, "Failed to authenticate")) - .addListener(ChannelFutureListener.CLOSE); + final ByteBuf msg = Commands.newError(-1, ServerError.AuthenticationError, "Failed to authenticate"); + writeAndFlushAndClose(msg); } } @@ -589,10 +587,10 @@ protected void handleAuthResponse(CommandAuthResponse authResponse) { }); } } catch (Exception e) { - String msg = "Unable to handleAuthResponse"; - LOG.warn("[{}] {} ", remoteAddress, msg, e); - ctx.writeAndFlush(Commands.newError(-1, ServerError.AuthenticationError, msg)) - .addListener(ChannelFutureListener.CLOSE); + String errorMsg = "Unable to handleAuthResponse"; + LOG.warn("[{}] {} ", remoteAddress, errorMsg, e); + final ByteBuf msg = Commands.newError(-1, ServerError.AuthenticationError, errorMsg); + writeAndFlushAndClose(msg); } } @@ -727,4 +725,12 @@ static boolean matchesHostAndPort(String expectedPrefix, String pulsarServiceUrl && pulsarServiceUrl.startsWith(expectedPrefix) && pulsarServiceUrl.startsWith(brokerHostPort, expectedPrefix.length()); } + + private void writeAndFlush(ByteBuf cmd) { + NettyChannelUtil.writeAndFlushWithVoidPromise(ctx, cmd); + } + + private void writeAndFlushAndClose(ByteBuf cmd) { + NettyChannelUtil.writeAndFlushWithClosePromise(ctx, cmd); + } } From 88f3dc40f076a7a9fec3849874fa5be0cde11ed8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=B2=20Boschi?= Date: Tue, 10 Jan 2023 11:21:40 +0100 Subject: [PATCH 3/4] style --- .../main/java/org/apache/pulsar/broker/service/ServerCnx.java | 1 - 1 file changed, 1 deletion(-) 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 aebb55d097e9a..c0357c14a2052 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 @@ -32,7 +32,6 @@ import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Strings; import io.netty.buffer.ByteBuf; -import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelOption; From d0e0aa4353ba4f99674f36a300ce4b91aec56609 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=B2=20Boschi?= Date: Tue, 10 Jan 2023 11:36:40 +0100 Subject: [PATCH 4/4] javadoc --- .../common/util/netty/NettyChannelUtil.java | 35 ++++++++++++++++--- 1 file changed, 30 insertions(+), 5 deletions(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/NettyChannelUtil.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/NettyChannelUtil.java index 2f7061ef4ca54..c466822e20576 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/NettyChannelUtil.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/netty/NettyChannelUtil.java @@ -21,18 +21,43 @@ import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelOutboundInvoker; +import io.netty.channel.VoidChannelPromise; /** * Contains utility methods for working with Netty Channels. */ -public class NettyChannelUtil { +public final class NettyChannelUtil { - public static void writeAndFlushWithVoidPromise(ChannelOutboundInvoker ctx, ByteBuf cmd) { - ctx.writeAndFlush(cmd, ctx.voidPromise()); + private NettyChannelUtil() { } - public static void writeAndFlushWithClosePromise(ChannelOutboundInvoker ctx, ByteBuf cmd) { - ctx.writeAndFlush(cmd).addListener(ChannelFutureListener.CLOSE); + /** + * Write and flush the message to the channel. + * + * The promise is an instance of {@link VoidChannelPromise} that properly propagates exceptions up to the pipeline. + * Netty has many ad-hoc optimization if the promise is an instance of {@link VoidChannelPromise}. + * Lastly, it reduces pollution of useless {@link io.netty.channel.ChannelPromise} objects created + * by the default write and flush method {@link ChannelOutboundInvoker#writeAndFlush(Object)}. + * See https://stackoverflow.com/q/54169262 and https://stackoverflow.com/a/9030420 for more details. + * + * @param ctx channel's context + * @param msg buffer to write in the channel + */ + public static void writeAndFlushWithVoidPromise(ChannelOutboundInvoker ctx, ByteBuf msg) { + ctx.writeAndFlush(msg, ctx.voidPromise()); + } + + /** + * Write and flush the message to the channel and the close the channel. + * + * This method is particularly helpful when the connection is in an invalid state + * and therefore a new connection must be created to continue. + * + * @param ctx channel's context + * @param msg buffer to write in the channel + */ + public static void writeAndFlushWithClosePromise(ChannelOutboundInvoker ctx, ByteBuf msg) { + ctx.writeAndFlush(msg).addListener(ChannelFutureListener.CLOSE); } }