From 706e4dde50c39dfd0ff6a01b6c44681d5a390ea3 Mon Sep 17 00:00:00 2001
From: Ning Yu <ningyu@automq.com>
Date: Mon, 5 Feb 2024 11:33:13 +0800
Subject: [PATCH] fix: register metrics after starting the wal

Signed-off-by: Ning Yu <ningyu@automq.com>
---
 .../automq/stream/s3/wal/BlockWALService.java    | 16 ++++++++++++----
 1 file changed, 12 insertions(+), 4 deletions(-)

diff --git a/s3stream/src/main/java/com/automq/stream/s3/wal/BlockWALService.java b/s3stream/src/main/java/com/automq/stream/s3/wal/BlockWALService.java
index 52eaa8601..91490cfba 100644
--- a/s3stream/src/main/java/com/automq/stream/s3/wal/BlockWALService.java
+++ b/s3stream/src/main/java/com/automq/stream/s3/wal/BlockWALService.java
@@ -290,7 +290,11 @@ public WriteAheadLog start() throws IOException {
         walHeaderReady(header);
 
         started.set(true);
+        LOGGER.info("block WAL service started, cost: {} ms", stopWatch.getTime(TimeUnit.MILLISECONDS));
+        return this;
+    }
 
+    private void registerMetrics() {
         S3StreamMetricsManager.registerDeltaWalOffsetSupplier(() -> {
             try {
                 return this.getCurrentStartOffset();
@@ -299,9 +303,6 @@ public WriteAheadLog start() throws IOException {
                 return 0L;
             }
         }, () -> walHeader.getFlushedTrimOffset());
-
-        LOGGER.info("block WAL service started, cost: {} ms", stopWatch.getTime(TimeUnit.MILLISECONDS));
-        return this;
     }
 
     private long getCurrentStartOffset() {
@@ -480,7 +481,14 @@ public CompletableFuture<Void> reset() {
             slidingWindowService.start(walHeader.getAtomicSlidingWindowMaxLength(), recoveryCompleteOffset);
         }
         LOGGER.info("reset sliding window to offset: {}", recoveryCompleteOffset);
-        return trim(recoveryCompleteOffset - 1, true).thenRun(() -> resetFinished.set(true));
+        CompletableFuture<Void> cf = trim(recoveryCompleteOffset - 1, true)
+            .thenRun(() -> resetFinished.set(true));
+
+        if (!recoveryMode) {
+            // Only register metrics when not in recovery mode
+            return cf.thenRun(this::registerMetrics);
+        }
+        return cf;
     }
 
     @Override