[fix] Run messageReadStats metrics registerFailedEvent on netty threa…

…ds in PartitionLog
streamnative · Sep 22, 2023 · 1a808ad · 1a808ad
1 parent 014340d
commit 1a808ad
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);