This is an automated email from the ASF dual-hosted git repository.
yubiao 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 faee1233d8b [improve] [broker] improve read entry error log for
troubleshooting (#21169)
faee1233d8b is described below
commit faee1233d8b9dab229d8e4c67fdd28c307da2ed1
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 +-
.../impl/SnapshotSegmentAbortedTxnProcessorImpl.java | 6 ++++++
8 files changed, 55 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 2e066863939..3631feccca9 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
@@ -816,6 +816,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);
@@ -1528,6 +1533,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 89c18f4b834..4991080d38c 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
@@ -1263,6 +1263,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 944674f6862..1fdf6939506 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 81f5e3c1f32..5a1462abeb7 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
@@ -2824,6 +2824,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"));
@@ -2922,6 +2928,12 @@ public class PersistentTopicsBase extends AdminResource {
}
}
}
+
+ @Override
+ public String toString() {
+ return String.format("Topic [{}] internal get
message by id",
+ PersistentTopicsBase.this.topicName);
+ }
}, null);
return results;
});
@@ -3088,6 +3100,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 ccf70eecec3..0b341524339 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
@@ -529,6 +529,12 @@ public abstract 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 0c77c7662fd..3487aa9fb2a 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
@@ -852,6 +852,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 85c1d98e047..22459b2f615 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
@@ -3040,7 +3040,7 @@ public class PersistentTopic extends AbstractTopic
implements Topic, AddEntryCal
// if AutoSkipNonRecoverableData is set to true, just return
true here.
return true;
} else {
- 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) {
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java
index fc92754d300..258a253c12e 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java
@@ -322,6 +322,12 @@ public class SnapshotSegmentAbortedTxnProcessorImpl
implements AbortedTxnProcess
hasInvalidIndex.set(true);
}
}
+
+ @Override
+ public String toString() {
+ return
String.format("Transaction buffer [{}] recover from snapshot",
+
SnapshotSegmentAbortedTxnProcessorImpl.this.topic.getName());
+ }
}, null);
});
openManagedLedgerAndHandleSegmentsFuture.complete(null);
