You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2024/12/16 12:14:12 UTC
(pulsar) 01/07: [improve][client] Print consumer stats log if prefetched messages are not zero (#23698)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8f9d7962ae25aff78e931a394f544d9f4b1052ea
Author: Penghui Li <pe...@apache.org>
AuthorDate: Thu Dec 12 21:32:20 2024 +0800
[improve][client] Print consumer stats log if prefetched messages are not zero (#23698)
(cherry picked from commit c52af1d5c733ec05b40fc72e63a202f20d25603b)
---
.../org/apache/pulsar/client/impl/ConsumerStatsRecorderImpl.java | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerStatsRecorderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerStatsRecorderImpl.java
index 8dfc0af8e1d..5cbbcc44298 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerStatsRecorderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerStatsRecorderImpl.java
@@ -146,15 +146,16 @@ public class ConsumerStatsRecorderImpl implements ConsumerStatsRecorder {
receivedMsgsRate = currentNumMsgsReceived / elapsed;
receivedBytesRate = currentNumBytesReceived / elapsed;
+ int prefetchQueueSize = consumerImpl.incomingMessages.size();
if ((currentNumMsgsReceived | currentNumBytesReceived | currentNumReceiveFailed | currentNumAcksSent
- | currentNumAcksFailed) != 0) {
+ | currentNumAcksFailed | prefetchQueueSize) != 0) {
log.info(
"[{}] [{}] [{}] Prefetched messages: {} --- "
+ "Consume throughput received: {} msgs/s --- {} Mbit/s --- "
+ "Ack sent rate: {} ack/s --- " + "Failed messages: {} --- batch messages: {} ---"
+ "Failed acks: {}",
consumerImpl.getTopic(), consumerImpl.getSubscription(), consumerImpl.consumerName,
- consumerImpl.incomingMessages.size(), THROUGHPUT_FORMAT.format(receivedMsgsRate),
+ prefetchQueueSize, THROUGHPUT_FORMAT.format(receivedMsgsRate),
THROUGHPUT_FORMAT.format(receivedBytesRate * 8 / 1024 / 1024),
THROUGHPUT_FORMAT.format(currentNumAcksSent / elapsed), currentNumReceiveFailed,
currentNumBatchReceiveFailed, currentNumAcksFailed);