From 8845a7739b7ec00431d360a9e58d7b7967ec1f82 Mon Sep 17 00:00:00 2001
From: zymap <zhangyong1025.zy@gmail.com>
Date: Wed, 6 Mar 2024 15:31:17 +0800
Subject: [PATCH] Fix the bookie doesn't transition to writable ---

### Motivation

When bookie started with readonly, and the disk usage is
small than the warning threshold, it won't transition to
write mode. Notify the disk writable when init the bookie
---
 .../bookkeeper/bookie/LedgerDirsMonitor.java  | 14 ++++++++--
 .../bookie/LedgerDirsManagerTest.java         | 26 +++++++++++++++++++
 2 files changed, 38 insertions(+), 2 deletions(-)

diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerDirsMonitor.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerDirsMonitor.java
index 7ad8ba1e0c5..21eec4c088b 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerDirsMonitor.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerDirsMonitor.java
@@ -66,7 +66,7 @@ public LedgerDirsMonitor(final ServerConfiguration conf,
         this.dirsManagers = dirsManagers;
     }
 
-    private void check(final LedgerDirsManager ldm) {
+    void check(final LedgerDirsManager ldm, boolean notifyDiskWritable) {
         final ConcurrentMap<File, Float> diskUsages = ldm.getDiskUsages();
         boolean someDiskFulled = false;
         boolean highPriorityWritesAllowed = true;
@@ -78,6 +78,11 @@ private void check(final LedgerDirsManager ldm) {
             for (File dir : writableDirs) {
                 try {
                     diskUsages.put(dir, diskChecker.checkDir(dir));
+                    if (notifyDiskWritable) {
+                        for (LedgerDirsListener listener : ldm.getListeners()) {
+                            listener.diskWritable(dir);
+                        }
+                    }
                 } catch (DiskErrorException e) {
                     LOG.error("Ledger directory {} failed on disk checking : ", dir, e);
                     // Notify disk failure to all listeners
@@ -193,7 +198,11 @@ private void check(final LedgerDirsManager ldm) {
     }
 
     private void check() {
-        dirsManagers.forEach(this::check);
+        check(false);
+    }
+
+    private void check(boolean notifyDiskWritable) {
+        dirsManagers.forEach(ldm -> check(ldm, notifyDiskWritable));
     }
 
     /**
@@ -207,6 +216,7 @@ private void check() {
      */
     public void init() throws DiskErrorException, NoWritableLedgerDirException {
         checkDirs();
+        check(true);
     }
 
     // start the daemon for disk monitoring
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/LedgerDirsManagerTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/LedgerDirsManagerTest.java
index d6e01cec457..ffcd1a493cb 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/LedgerDirsManagerTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/LedgerDirsManagerTest.java
@@ -528,6 +528,32 @@ public void testValidateLwmThreshold() {
         new LedgerDirsMonitor(configuration, mockDiskChecker, Collections.singletonList(dirsManager));
     }
 
+    @Test
+    public void testBookieStartWithReadonlyAndDiskIsLowerThanThreshold() throws Exception {
+        File tmpDir1 = createTempDir("bkTest", ".dir");
+        File curDir1 = BookieImpl.getCurrentDirectory(tmpDir1);
+        BookieImpl.checkDirectoryStructure(curDir1);
+
+        final ServerConfiguration configuration = TestBKConfiguration.newServerConfiguration();
+        configuration.setDiskUsageThreshold(0.95f);
+        configuration.setDiskUsageWarnThreshold(0.80f);
+        configuration.setDiskLowWaterMarkUsageThreshold(0.80f);
+        configuration.setLedgerDirNames(new String[]{tmpDir1.toString()});
+
+        MockDiskChecker diskChecker = new MockDiskChecker(configuration.getDiskUsageThreshold(), configuration.getDiskUsageWarnThreshold());
+
+        LedgerDirsManager ldm = new LedgerDirsManager(configuration, configuration.getLedgerDirs(), diskChecker, statsLogger);
+        LedgerDirsMonitor ldmMonitor = new LedgerDirsMonitor(configuration, diskChecker, Collections.singletonList(ldm));
+        MockLedgerDirsListener mockLedgerDirsListener = new MockLedgerDirsListener();
+        mockLedgerDirsListener.readOnly = true;
+        ldm.addLedgerDirsListener(mockLedgerDirsListener);
+        HashMap<File, Float> usageMap = new HashMap<>();
+        usageMap.put(curDir1, 0.75f);
+        diskChecker.setUsageMap(usageMap);
+        ldmMonitor.init();
+        assertFalse(mockLedgerDirsListener.readOnly);
+    }
+
     private void setUsageAndThenVerify(File dir1, float dir1Usage, File dir2, float dir2Usage,
             MockDiskChecker mockDiskChecker, MockLedgerDirsListener mockLedgerDirsListener, boolean verifyReadOnly)
             throws InterruptedException {