fix(wal): shutdown the pollBlockScheduler during shutdown

Signed-off-by: Ning Yu <[email protected]>

s3stream/src/main/java/com/automq/stream/s3/wal/SlidingWindowService.java

@@ -94,7 +94,14 @@ public class SlidingWindowService {
      */
     private Block currentBlock;
 
+    /**
+     * The thread pool for write operations.
+     */
     private ExecutorService ioExecutor;
+    /**
+     * The scheduler for polling blocks and sending them to @{@link #ioExecutor}.
+     */
+    private ScheduledExecutorService pollBlockScheduler;
 
     /**
      * The last time when a batch of blocks is written to the disk.
@@ -121,7 +128,7 @@ public void start(AtomicLong windowMaxLength, long windowStartOffset) {
         this.windowCoreData = new WindowCoreData(windowMaxLength, windowStartOffset, windowStartOffset);
         this.ioExecutor = Threads.newFixedThreadPoolWithMonitor(ioThreadNums,
             "block-wal-io-thread", false, LOGGER);
-        ScheduledExecutorService pollBlockScheduler = Threads.newSingleThreadScheduledExecutor(
+        this.pollBlockScheduler = Threads.newSingleThreadScheduledExecutor(
             ThreadUtils.createThreadFactory("wal-poll-block-thread-%d", false), LOGGER);
         pollBlockScheduler.scheduleAtFixedRate(this::tryWriteBlock, 0, minWriteIntervalNanos, TimeUnit.NANOSECONDS);
         initialized.set(true);
@@ -138,6 +145,7 @@ public boolean shutdown(long timeout, TimeUnit unit) {
 
         boolean gracefulShutdown;
         this.ioExecutor.shutdown();
+        this.pollBlockScheduler.shutdownNow();
         try {
             gracefulShutdown = this.ioExecutor.awaitTermination(timeout, unit);
         } catch (InterruptedException e) {