This is an automated email from the ASF dual-hosted git repository. eolivelli 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 c243d18 Issue #6054 - Adding more permits debug statements to better see changes to permits (#10217) c243d18 is described below commit c243d185261e964010a122b65f9c730bffbb2f0c Author: Devin Bost <devinb...@users.noreply.github.com> AuthorDate: Mon Apr 19 03:28:47 2021 -0600 Issue #6054 - Adding more permits debug statements to better see changes to permits (#10217) --- docker/build.sh | 2 +- .../org/apache/pulsar/broker/service/Consumer.java | 26 ++++++++++++++++++++++ .../broker/service/PulsarCommandSenderImpl.java | 4 ++-- .../PersistentDispatcherMultipleConsumers.java | 23 ++++++++++++++----- 4 files changed, 47 insertions(+), 8 deletions(-) diff --git a/docker/build.sh b/docker/build.sh index 944c35b..1379790 100755 --- a/docker/build.sh +++ b/docker/build.sh @@ -22,4 +22,4 @@ ROOT_DIR=$(git rev-parse --show-toplevel) cd $ROOT_DIR/docker mvn -f ../dashboard/pom.xml package -Pdocker -mvn package -Pdocker +mvn package -Pdocker diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java index 263d722..475fdc9 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java @@ -237,6 +237,11 @@ public class Consumer { if (entry != null) { int batchSize = batchSizes.getBatchSize(i); pendingAcks.put(entry.getLedgerId(), entry.getEntryId(), batchSize, 0); + if (log.isDebugEnabled()){ + log.debug("[{}-{}] Added {}:{} ledger entry with batchSize of {} to pendingAcks in" + + " broker.service.Consumer for consumerId: {}", + topicName, subscription, entry.getLedgerId(), entry.getEntryId(), batchSize, consumerId); + } } } } @@ -250,6 +255,11 @@ public class Consumer { // reduce permit and increment unackedMsg count with total number of messages in batch-msgs int ackedCount = batchIndexesAcks == null ? 0 : batchIndexesAcks.getTotalAckedIndexCount(); MESSAGE_PERMITS_UPDATER.addAndGet(this, ackedCount - totalMessages); + if (log.isDebugEnabled()){ + log.debug("[{}-{}] Added {} minus {} messages to MESSAGE_PERMITS_UPDATER in broker.service.Consumer" + + " for consumerId: {}; avgMessagesPerEntry is {}", + topicName, subscription, ackedCount, totalMessages, consumerId, tmpAvgMessagesPerEntry); + } incrementUnackedMessages(totalMessages); msgOut.recordMultipleEvents(totalMessages, totalBytes); msgOutCounter.add(totalMessages); @@ -506,6 +516,10 @@ public class Consumer { int oldPermits; if (!blockedConsumerOnUnackedMsgs) { oldPermits = MESSAGE_PERMITS_UPDATER.getAndAdd(this, additionalNumberOfMessages); + if (log.isDebugEnabled()) { + log.debug("[{}-{}] Added {} message permits in broker.service.Consumer before updating dispatcher " + + "for consumer", topicName, subscription, additionalNumberOfMessages, consumerId); + } subscription.consumerFlow(this, additionalNumberOfMessages); } else { oldPermits = PERMITS_RECEIVED_WHILE_CONSUMER_BLOCKED_UPDATER.getAndAdd(this, additionalNumberOfMessages); @@ -529,6 +543,10 @@ public class Consumer { int additionalNumberOfPermits = PERMITS_RECEIVED_WHILE_CONSUMER_BLOCKED_UPDATER.getAndSet(consumer, 0); // add newly flow permits to actual consumer.messagePermits MESSAGE_PERMITS_UPDATER.getAndAdd(consumer, additionalNumberOfPermits); + if (log.isDebugEnabled()){ + log.debug("[{}-{}] Added {} blocked permits to broker.service.Consumer for consumer", topicName, + subscription, additionalNumberOfPermits, consumerId); + } // dispatch pending permits to flow more messages: it will add more permits to dispatcher and consumer subscription.consumerFlow(consumer, additionalNumberOfPermits); } @@ -577,6 +595,10 @@ public class Consumer { lastAckedTimestamp = consumerStats.lastAckedTimestamp; lastConsumedTimestamp = consumerStats.lastConsumedTimestamp; MESSAGE_PERMITS_UPDATER.set(this, consumerStats.availablePermits); + if (log.isDebugEnabled()){ + log.debug("[{}-{}] Setting broker.service.Consumer's messagePermits to {} for consumer", topicName, + subscription, consumerStats.availablePermits, consumerId); + } unackedMessages = consumerStats.unackedMessages; blockedConsumerOnUnackedMsgs = consumerStats.blockedConsumerOnUnackedMsgs; AVG_MESSAGES_PER_ENTRY.set(this, consumerStats.avgMessagesPerEntry); @@ -754,6 +776,10 @@ public class Consumer { // if permitsReceivedWhileConsumerBlocked has been accumulated then pass it to Dispatcher to flow messages if (numberOfBlockedPermits > 0) { MESSAGE_PERMITS_UPDATER.getAndAdd(this, numberOfBlockedPermits); + if (log.isDebugEnabled()) { + log.debug("[{}-{}] Added {} blockedPermits to broker.service.Consumer's messagePermits for consumer {}", + topicName, subscription, numberOfBlockedPermits, consumerId); + } subscription.consumerFlow(this, numberOfBlockedPermits); } } 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 41705e9..dcc8458 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 @@ -284,8 +284,8 @@ public class PulsarCommandSenderImpl implements PulsarCommandSender { } if (log.isDebugEnabled()) { - log.debug("[{}-{}] Sending message to consumerId {}, msg id {}-{}", topicName, subscription, - consumerId, entry.getLedgerId(), entry.getEntryId()); + log.debug("[{}-{}] Sending message to consumerId {}, msg id {}-{} with batchSize {}", + topicName, subscription, consumerId, entry.getLedgerId(), entry.getEntryId(), batchSize); } int redeliveryCount = 0; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java index af22fa5..c2dc7d7 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java @@ -213,6 +213,11 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul redeliveryTracker.addIfAbsent(PositionImpl.get(ledgerId, entryId)); }); totalAvailablePermits -= consumer.getAvailablePermits(); + if (log.isDebugEnabled()) { + log.debug("[{}] Decreased totalAvailablePermits by {} in PersistentDispatcherMultipleConsumers. " + + "New dispatcher permit count is {}", name, consumer.getAvailablePermits(), + totalAvailablePermits); + } readMoreEntries(); } } else { @@ -232,8 +237,9 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul totalAvailablePermits += additionalNumberOfMessages; if (log.isDebugEnabled()) { - log.debug("[{}-{}] Trigger new read after receiving flow control message with permits {}", name, consumer, - totalAvailablePermits); + log.debug("[{}-{}] Trigger new read after receiving flow control message with permits {} " + + "after adding {} permits", name, consumer, + totalAvailablePermits, additionalNumberOfMessages); } readMoreEntries(); } @@ -494,8 +500,9 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul // round-robin dispatch batch size for this consumer int availablePermits = c.isWritable() ? c.getAvailablePermits() : 1; if (log.isDebugEnabled() && !c.isWritable()) { - log.debug("[{}-{}] consumer is not writable. dispatching only 1 message to {} ", topic.getName(), name, - c); + log.debug("[{}-{}] consumer is not writable. dispatching only 1 message to {}; " + + "availablePermits are {}", topic.getName(), name, + c, availablePermits); } int messagesForC = Math.min( Math.min(entriesToDispatch, availablePermits), @@ -524,7 +531,13 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul int msgSent = sendMessageInfo.getTotalMessages(); start += messagesForC; entriesToDispatch -= messagesForC; - TOTAL_AVAILABLE_PERMITS_UPDATER.addAndGet(this, -(msgSent - batchIndexesAcks.getTotalAckedIndexCount())); + TOTAL_AVAILABLE_PERMITS_UPDATER.addAndGet(this, + -(msgSent - batchIndexesAcks.getTotalAckedIndexCount())); + if (log.isDebugEnabled()){ + log.debug("[{}] Added -({} minus {}) permits to TOTAL_AVAILABLE_PERMITS_UPDATER in " + + "PersistentDispatcherMultipleConsumers", + name, msgSent, batchIndexesAcks.getTotalAckedIndexCount()); + } totalMessagesSent += sendMessageInfo.getTotalMessages(); totalBytesSent += sendMessageInfo.getTotalBytes(); }