This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.11 by this push:
new 6ee57c40477 [improve] [broker] improve read entry error log for
troubleshooting (#21169)
6ee57c40477 is described below
commit 6ee57c4047727fb614fd7243fe2322c1085cb596
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)
---
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 10 ++++++++++
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 6 ++++++
.../mledger/impl/ReadOnlyManagedLedgerImpl.java | 4 ++--
.../pulsar/broker/admin/impl/PersistentTopicsBase.java | 18 ++++++++++++++++++
.../service/persistent/PersistentReplicator.java | 6 ++++++
.../service/persistent/PersistentSubscription.java | 6 ++++++
.../broker/service/persistent/PersistentTopic.java | 2 +-
7 files changed, 49 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 edc00bf4722..cedc6c34fc3 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
@@ -751,6 +751,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);
@@ -1448,6 +1453,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 358f8ce665e..7e00e675f61 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
@@ -1239,6 +1239,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 465be8b01c5..6f4ca87e90c 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
@@ -2690,6 +2690,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"));
@@ -2776,6 +2782,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.
@@ -2945,6 +2957,12 @@ public class PersistentTopicsBase extends AdminResource {
public void readEntryFailed(ManagedLedgerException
exception, Object ctx) {
future.completeExceptionally(exception);
}
+
+ @Override
+ public String toString() {
+ return String.format("Topic [{}] internal
examine message async",
+ PersistentTopicsBase.this.topicName);
+ }
}, null);
return future;
} catch (ManagedLedgerException exception) {
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 eebda4f5556..3e1120ddc09 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 1a97b9f784a..e2e5c4aa053 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
@@ -844,6 +844,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 4a3e38c5395..cdd37f97952 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
@@ -2680,7 +2680,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();
