You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/09/11 04:11:31 UTC
[pulsar] branch branch-3.0 updated: [improve] [broker] Improve logs for troubleshooting (#21141)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new 78b26eb4231 [improve] [broker] Improve logs for troubleshooting (#21141)
78b26eb4231 is described below
commit 78b26eb4231f086c25065c24e6e88e52151a09ba
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Thu Sep 7 03:15:22 2023 +0800
[improve] [broker] Improve logs for troubleshooting (#21141)
---
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 26 +++++++++++++---------
.../mledger/impl/NonDurableCursorImpl.java | 8 +++++--
.../mledger/impl/NonDurableCursorTest.java | 3 ++-
.../broker/service/persistent/PersistentTopic.java | 8 +++----
4 files changed, 27 insertions(+), 18 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index 965e078bf02..2e066863939 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -509,7 +509,7 @@ public class ManagedCursorImpl implements ManagedCursor {
callback.operationComplete();
} else {
// Need to proceed and read the last entry in the specified ledger to find out the last position
- log.info("[{}] Consumer {} meta-data recover from ledger {}", ledger.getName(), name,
+ log.info("[{}] Cursor {} meta-data recover from ledger {}", ledger.getName(), name,
info.getCursorsLedgerId());
recoverFromLedger(info, callback);
}
@@ -529,16 +529,16 @@ public class ManagedCursorImpl implements ManagedCursor {
long ledgerId = info.getCursorsLedgerId();
OpenCallback openCallback = (rc, lh, ctx) -> {
if (log.isInfoEnabled()) {
- log.info("[{}] Opened ledger {} for consumer {}. rc={}", ledger.getName(), ledgerId, name, rc);
+ log.info("[{}] Opened ledger {} for cursor {}. rc={}", ledger.getName(), ledgerId, name, rc);
}
if (isBkErrorNotRecoverable(rc)) {
- log.error("[{}] Error opening metadata ledger {} for consumer {}: {}", ledger.getName(), ledgerId, name,
+ log.error("[{}] Error opening metadata ledger {} for cursor {}: {}", ledger.getName(), ledgerId, name,
BKException.getMessage(rc));
// Rewind to oldest entry available
initialize(getRollbackPosition(info), Collections.emptyMap(), Collections.emptyMap(), callback);
return;
} else if (rc != BKException.Code.OK) {
- log.warn("[{}] Error opening metadata ledger {} for consumer {}: {}", ledger.getName(), ledgerId, name,
+ log.warn("[{}] Error opening metadata ledger {} for cursor {}: {}", ledger.getName(), ledgerId, name,
BKException.getMessage(rc));
callback.operationFailed(new ManagedLedgerException(BKException.getMessage(rc)));
return;
@@ -548,7 +548,7 @@ public class ManagedCursorImpl implements ManagedCursor {
long lastEntryInLedger = lh.getLastAddConfirmed();
if (lastEntryInLedger < 0) {
- log.warn("[{}] Error reading from metadata ledger {} for consumer {}: No entries in ledger",
+ log.warn("[{}] Error reading from metadata ledger {} for cursor {}: No entries in ledger",
ledger.getName(), ledgerId, name);
// Rewind to last cursor snapshot available
initialize(getRollbackPosition(info), Collections.emptyMap(), cursorProperties, callback);
@@ -560,13 +560,13 @@ public class ManagedCursorImpl implements ManagedCursor {
log.debug("[{}} readComplete rc={} entryId={}", ledger.getName(), rc1, lh1.getLastAddConfirmed());
}
if (isBkErrorNotRecoverable(rc1)) {
- log.error("[{}] Error reading from metadata ledger {} for consumer {}: {}", ledger.getName(),
+ log.error("[{}] Error reading from metadata ledger {} for cursor {}: {}", ledger.getName(),
ledgerId, name, BKException.getMessage(rc1));
// Rewind to oldest entry available
initialize(getRollbackPosition(info), Collections.emptyMap(), cursorProperties, callback);
return;
} else if (rc1 != BKException.Code.OK) {
- log.warn("[{}] Error reading from metadata ledger {} for consumer {}: {}", ledger.getName(),
+ log.warn("[{}] Error reading from metadata ledger {} for cursor {}: {}", ledger.getName(),
ledgerId, name, BKException.getMessage(rc1));
callback.operationFailed(createManagedLedgerException(rc1));
@@ -2454,8 +2454,12 @@ public class ManagedCursorImpl implements ManagedCursor {
@Override
public synchronized String toString() {
- return MoreObjects.toStringHelper(this).add("ledger", ledger.getName()).add("name", name)
- .add("ackPos", markDeletePosition).add("readPos", readPosition).toString();
+ return MoreObjects.toStringHelper(this)
+ .add("ledger", ledger.getName())
+ .add("name", name)
+ .add("ackPos", markDeletePosition)
+ .add("readPos", readPosition)
+ .toString();
}
@Override
@@ -3069,7 +3073,7 @@ public class ManagedCursorImpl implements ManagedCursor {
if (shouldCloseLedger(lh1)) {
if (log.isDebugEnabled()) {
- log.debug("[{}] Need to create new metadata ledger for consumer {}", ledger.getName(), name);
+ log.debug("[{}] Need to create new metadata ledger for cursor {}", ledger.getName(), name);
}
startCreatingNewMetadataLedger();
}
@@ -3154,7 +3158,7 @@ public class ManagedCursorImpl implements ManagedCursor {
@Override
public void operationFailed(MetaStoreException e) {
- log.warn("[{}] Failed to update consumer {}", ledger.getName(), name, e);
+ log.warn("[{}] Failed to update cursor metadata {}", ledger.getName(), name, e);
// it means it failed to switch the newly created ledger so, it should be
// deleted to prevent leak
deleteLedgerAsync(lh).thenRun(() -> callback.operationFailed(e));
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java
index 9d2829b1707..51e56158cad 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java
@@ -138,8 +138,12 @@ public class NonDurableCursorImpl extends ManagedCursorImpl {
@Override
public synchronized String toString() {
- return MoreObjects.toStringHelper(this).add("ledger", ledger.getName()).add("ackPos", markDeletePosition)
- .add("readPos", readPosition).toString();
+ return MoreObjects.toStringHelper(this)
+ .add("ledger", ledger.getName())
+ .add("cursor", getName())
+ .add("ackPos", markDeletePosition)
+ .add("readPos", readPosition)
+ .toString();
}
private static final Logger log = LoggerFactory.getLogger(NonDurableCursorImpl.class);
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorTest.java
index 1ad3f5f8de6..1e1f7df0a46 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorTest.java
@@ -80,7 +80,8 @@ public class NonDurableCursorTest extends MockedBookKeeperTestCase {
entries.forEach(Entry::release);
// Test string representation
- assertEquals(c1.toString(), "NonDurableCursorImpl{ledger=my_test_ledger, ackPos=3:-1, readPos=3:1}");
+ assertEquals(c1.toString(), "NonDurableCursorImpl{ledger=my_test_ledger, cursor="
+ + c1.getName() + ", ackPos=3:-1, readPos=3:1}");
}
@Test(timeOut = 20000)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 948c6710447..85c1d98e047 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -901,8 +901,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
consumer.close();
} catch (BrokerServiceException e) {
if (e instanceof ConsumerBusyException) {
- log.warn("[{}][{}] Consumer {} {} already connected",
- topic, subscriptionName, consumerId, consumerName);
+ log.warn("[{}][{}] Consumer {} {} already connected: {}",
+ topic, subscriptionName, consumerId, consumerName, e.getMessage());
} else if (e instanceof SubscriptionBusyException) {
log.warn("[{}][{}] {}", topic, subscriptionName, e.getMessage());
}
@@ -932,8 +932,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
decrementUsageCount();
if (ex.getCause() instanceof ConsumerBusyException) {
- log.warn("[{}][{}] Consumer {} {} already connected", topic, subscriptionName, consumerId,
- consumerName);
+ log.warn("[{}][{}] Consumer {} {} already connected: {}", topic, subscriptionName, consumerId,
+ consumerName, ex.getCause().getMessage());
Consumer consumer = null;
try {
consumer = subscriptionFuture.isDone() ? getActiveConsumer(subscriptionFuture.get()) : null;