poorbarcode commented on code in PR #20568:
URL: https://github.com/apache/pulsar/pull/20568#discussion_r1231250086
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java:
##########
@@ -361,6 +361,15 @@ public Future<Void> sendMessages(final List<? extends
Entry> entries, EntryBatch
msgOutCounter.add(totalMessages);
bytesOutCounter.add(totalBytes);
chunkedMessageRate.recordMultipleEvents(totalChunkedMessages,
0);
+ } else {
+ log.warn("[{}-{}] Sent messages to client fail by IO
exception[{}], these messages(messages count:"
+ + " {}) will be redelivered after the
heartbeat check fails. If the next heartbeat"
+ + " check is successful, these messages will
be stuck until the client reconnect"
+ + " or the topic is reloaded. Consumer: {}",
+ topicName, subscription, status.cause() == null ? "" :
status.cause().getMessage(),
+ totalMessages, this.toString(), status.cause());
+ // If the health check fail, this connection will be closed.
+ cnx.healthCheckManually();
Review Comment:
It is a good suggestion. The method `checkConnectionLiveness` was called
when a new consumer registered, it always calls `sendPing`. Since sending
messages to the client is a high-frequency event. So if a connection is
inactive, this results in a large number of tasks being pushed into Netty's
task pool. We should wrap the method `checkConnectionLiveness` as a new method
to limit the number of tasks. such as
```java
public void healthCheckManually( ) {
if (canSendPingNow()) {
checkConnectionLiveness();
}
}
```
But the wrap method is not needed to switch thread to the netty thread, it
already runs in the netty thread, so calling `sendPing` directly is better than
calling `checkConnectionLiveness`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
