This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.10 by this push:
new 84902ac74b6 [improve] [broker] improve read entry error log for
troubleshooting (#21169)
84902ac74b6 is described below
commit 84902ac74b6d0bc91a3f886e35609e160b48a776
Author: fengyubiao <[email protected]>
AuthorDate: Wed Sep 13 17:30:57 2023 +0800
[improve] [broker] improve read entry error log for troubleshooting (#21169)
(cherry picked from commit 65706c6ffa737f946ad9a1bfdb6ff70fa66c0415)
---
.../apache/bookkeeper/mledger/impl/ManagedCursorImpl.java | 10 ++++++++++
.../apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java | 6 ++++++
.../bookkeeper/mledger/impl/ReadOnlyManagedLedgerImpl.java | 4 ++--
.../pulsar/broker/admin/impl/PersistentTopicsBase.java | 12 ++++++++++++
.../broker/service/persistent/PersistentReplicator.java | 6 ++++++
.../broker/service/persistent/PersistentSubscription.java | 6 ++++++
.../pulsar/broker/service/persistent/PersistentTopic.java | 2 +-
7 files changed, 43 insertions(+), 3 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 eb6fb6fe3a3..35da0f33f5d 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
@@ -750,6 +750,11 @@ public class ManagedCursorImpl implements ManagedCursor {
result.entry = entry;
counter.countDown();
}
+
+ @Override
+ public String toString() {
+ return String.format("Cursor [{}] get Nth entry",
ManagedCursorImpl.this);
+ }
}, null);
counter.await(ledger.getConfig().getMetadataOperationsTimeoutSeconds(),
TimeUnit.SECONDS);
@@ -1426,6 +1431,11 @@ public class ManagedCursorImpl implements ManagedCursor {
callback.readEntriesFailed(exception.get(), ctx);
}
}
+
+ @Override
+ public String toString() {
+ return String.format("Cursor [{}] async replay entries",
ManagedCursorImpl.this);
+ }
};
positions.stream().filter(position ->
!alreadyAcknowledgedPositions.contains(position))
diff --git
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index b5b5cd21ff8..9a6a45550f9 100644
---
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -1236,6 +1236,12 @@ public class ManagedLedgerImpl implements ManagedLedger,
CreateCallback {
log.error("Error read entry for position {}", nextPos,
exception);
future.completeExceptionally(exception);
}
+
+ @Override
+ public String toString() {
+ return String.format("ML [{}] get earliest message publish
time of pos",
+ ManagedLedgerImpl.this.name);
+ }
}, null);
return future;
diff --git
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadOnlyManagedLedgerImpl.java
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadOnlyManagedLedgerImpl.java
index 214c3afe1bc..f1ef7b04857 100644
---
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadOnlyManagedLedgerImpl.java
+++
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadOnlyManagedLedgerImpl.java
@@ -143,8 +143,8 @@ public class ReadOnlyManagedLedgerImpl extends
ManagedLedgerImpl {
this.getLedgerHandle(position.getLedgerId())
.thenAccept((ledger) -> asyncReadEntry(ledger, position,
callback, ctx))
.exceptionally((ex) -> {
- log.error("[{}] Error opening ledger for reading at
position {} - {}", this.name, position,
- ex.getMessage());
+ log.error("[{}] Error opening ledger for reading at
position {} - {}. Op: {}", this.name,
+ position, ex.getMessage(), callback);
callback.readEntryFailed(ManagedLedgerException.getManagedLedgerException(ex.getCause()),
ctx);
return null;
});
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index f4e5caacded..1ec1c3d8355 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -2558,6 +2558,12 @@ public class PersistentTopicsBase extends AdminResource {
}
}
}
+
+ @Override
+ public String toString() {
+ return String.format("Topic [{}] get entry batch size",
+ PersistentTopicsBase.this.topicName);
+ }
}, null);
} catch (NullPointerException npe) {
batchSizeFuture.completeExceptionally(new
RestException(Status.NOT_FOUND, "Message not found"));
@@ -2644,6 +2650,12 @@ public class PersistentTopicsBase extends AdminResource {
}
}
}
+
+ @Override
+ public String toString() {
+ return String.format("Topic [{}] internal
get message by id",
+
PersistentTopicsBase.this.topicName);
+ }
}, null);
}).exceptionally(ex -> {
// If the exception is not redirect exception we need to
log it.
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
index adfae64489d..1f47cdecb16 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
@@ -653,6 +653,12 @@ public class PersistentReplicator extends
AbstractReplicator
public void readEntryComplete(Entry entry, Object ctx) {
future.complete(entry);
}
+
+ @Override
+ public String toString() {
+ return String.format("Replication [{}] peek Nth message",
+ PersistentReplicator.this.producer.getProducerName());
+ }
}, null);
return future;
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
index a14a5108dff..829e2d14cad 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
@@ -730,6 +730,12 @@ public class PersistentSubscription extends
AbstractSubscription implements Subs
public void readEntryComplete(Entry entry, Object ctx) {
future.complete(entry);
}
+
+ @Override
+ public String toString() {
+ return String.format("Subscription [{}-{}] async replay
entries", PersistentSubscription.this.topicName,
+ PersistentSubscription.this.subName);
+ }
}, null);
return future;
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 14074fc280b..ebaee40eeec 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
@@ -2663,7 +2663,7 @@ public class PersistentTopic extends AbstractTopic
implements Topic, AddEntryCal
(int) (messageTTLInSeconds *
MESSAGE_EXPIRY_THRESHOLD), entryTimestamp);
}
} catch (Exception e) {
- log.warn("[{}] Error while getting the oldest message", topic, e);
+ log.warn("[{}] [{}] Error while getting the oldest message",
topic, cursor.toString(), e);
} finally {
if (entry != null) {
entry.release();
