[improve] [broker] improve read entry error log for troubleshooting (apache#21169)

(cherry picked from commit 65706c6)

Author: poorbarcode

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

@@ -750,6 +750,11 @@ public void readEntryComplete(Entry entry, Object ctx) {
                 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 synchronized void readEntryFailed(ManagedLedgerException mle, Object ctx)
                     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))

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

@@ -1236,6 +1236,12 @@ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
                 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;

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadOnlyManagedLedgerImpl.java

@@ -143,8 +143,8 @@ public void asyncReadEntry(PositionImpl position, AsyncCallbacks.ReadEntryCallba
             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;
                     });

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

@@ -2558,6 +2558,12 @@ public void readEntryComplete(Entry entry, Object ctx) {
                             }
                         }
                     }
+
+                    @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 void readEntryComplete(Entry entry, Object ctx) {
                                         }
                                     }
                                 }
+
+                                @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.

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java

@@ -653,6 +653,12 @@ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
             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;

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

@@ -730,6 +730,12 @@ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
             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;

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -2663,7 +2663,7 @@ public boolean isOldestMessageExpired(ManagedCursor cursor, int messageTTLInSeco
                         (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();