You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by ji...@apache.org on 2022/07/29 12:09:22 UTC
[pulsar] branch branch-2.7 updated: [Broker/Client] Close connection if a ping or pong message cannot be sent (#15382)
This is an automated email from the ASF dual-hosted git repository.
jianghaiting pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.7 by this push:
new caf3fac4798 [Broker/Client] Close connection if a ping or pong message cannot be sent (#15382)
caf3fac4798 is described below
commit caf3fac4798c127d3056498ce2e3d807d22871cb
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Apr 29 17:07:11 2022 +0300
[Broker/Client] Close connection if a ping or pong message cannot be sent (#15382)
* [Broker/Client] Close connection if a ping message cannot be sent
- the connection should be closed if a ping message cannot be sent
* Handle write errors for pong messages
(cherry picked from commit 2ddef95f31ce37486f3f76b4d59730361a77bf6e)
---
.../apache/pulsar/common/protocol/PulsarHandler.java | 18 ++++++++++++++++--
1 file changed, 16 insertions(+), 2 deletions(-)
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarHandler.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarHandler.java
index ea16251d40e..bafe4a5d442 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarHandler.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarHandler.java
@@ -77,7 +77,14 @@ public abstract class PulsarHandler extends PulsarDecoder {
if (log.isDebugEnabled()) {
log.debug("[{}] Replying back to ping message", ctx.channel());
}
- ctx.writeAndFlush(Commands.newPong());
+ ctx.writeAndFlush(Commands.newPong())
+ .addListener(future -> {
+ if (!future.isSuccess()) {
+ log.warn("[{}] Forcing connection to close since cannot send a pong message.",
+ ctx.channel(), future.cause());
+ ctx.close();
+ }
+ });
}
@Override
@@ -104,7 +111,14 @@ public abstract class PulsarHandler extends PulsarDecoder {
log.debug("[{}] Sending ping message", ctx.channel());
}
waitingForPingResponse = true;
- ctx.writeAndFlush(Commands.newPing());
+ ctx.writeAndFlush(Commands.newPing())
+ .addListener(future -> {
+ if (!future.isSuccess()) {
+ log.warn("[{}] Forcing connection to close since cannot send a ping message.",
+ ctx.channel(), future.cause());
+ ctx.close();
+ }
+ });
} else {
if (log.isDebugEnabled()) {
log.debug("[{}] Peer doesn't support keep-alive", ctx.channel());