[fix] Run messageReadStats metrics registerFailedEvent execute on net…

…ty threa… (#2007) Fix: follow to #2006
streamnative · Oct 8, 2023 · db4a9ad · db4a9ad
1 parent 014340d
commit db4a9ad
Showing 1 changed file with 4 additions and 2 deletions.
diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/PartitionLog.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/PartitionLog.java
@@ -884,8 +884,10 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                 if (exception instanceof ManagedLedgerException.ManagedLedgerFencedException) {
                     invalidateCacheOnTopic.accept(fullPartitionName);
                 }
-                messageReadStats.registerFailedEvent(
-                        MathUtils.elapsedNanos(startReadingMessagesNanos), TimeUnit.NANOSECONDS);
+                long failedLatencyNanos = MathUtils.elapsedNanos(startReadingMessagesNanos);
+                eventExecutor.execute(() -> {
+                    messageReadStats.registerFailedEvent(failedLatencyNanos, TimeUnit.NANOSECONDS);
+                });
                 readFuture.completeExceptionally(exception);
             }
         }, null, PositionImpl.LATEST);