Add diskUnderWarnThreshold check event to prevent gcThread suspended forever

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorage.java

@@ -208,6 +208,15 @@ public void diskAlmostFull(File disk) {
                 }
             }
 
+            @Override
+            public void diskUnderWarnThreshold(File disk) {
+                if (gcThread.isForceGCAllowWhenNoSpace()) {
+                    gcThread.disableForceGC();
+                } else {
+                    gcThread.resumeMajorGC();
+                }
+            }
+
             @Override
             public void diskFull(File disk) {
                 if (gcThread.isForceGCAllowWhenNoSpace) {

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerDirsManager.java

@@ -47,6 +47,7 @@ public class LedgerDirsManager {
     private static final Logger LOG = LoggerFactory.getLogger(LedgerDirsManager.class);
 
     private volatile List<File> filledDirs;
+    private volatile List<File> warnDirs;
     private final List<File> ledgerDirectories;
     private volatile List<File> writableLedgerDirectories;
     private final List<LedgerDirsListener> listeners;
@@ -71,6 +72,7 @@ public LedgerDirsManager(ServerConfiguration conf, File[] dirs, DiskChecker disk
         }
         this.writableLedgerDirectories = new ArrayList<File>(ledgerDirectories);
         this.filledDirs = new ArrayList<File>();
+        this.warnDirs = new ArrayList<File>();
         this.listeners = new ArrayList<LedgerDirsListener>();
         this.entryLogSize = conf.getEntryLogSizeLimit();
         this.minUsableSizeForIndexFileCreation = conf.getMinUsableSizeForIndexFileCreation();
@@ -233,6 +235,13 @@ public List<File> getFullFilledLedgerDirs() {
         return filledDirs;
     }
 
+    /**
+     * @return warn threshold ledger dirs.
+     */
+    public List<File> getWarnLedgerDirs() {
+        return warnDirs;
+    }
+
     /**
      * Get dirs, which are full more than threshold.
      */
@@ -290,6 +299,39 @@ public void addToWritableDirs(File dir, boolean underWarnThreshold) {
         }
     }
 
+    /**
+     * Add the dir to warn dirs list.
+     */
+    @VisibleForTesting
+    public void addToWarnDirs(File dir) {
+        if (!warnDirs.contains(dir)) {
+            LOG.warn(dir + " usages is above warn threshold. Adding it to warn dirs list");
+            List<File> updatedFilledDirs = new ArrayList<File>(warnDirs);
+            updatedFilledDirs.add(dir);
+            warnDirs = updatedFilledDirs;
+            for (LedgerDirsListener listener : listeners) {
+                listener.diskAlmostFull(dir);
+            }
+        }
+    }
+
+
+    /**
+     * Remove warn dirs list.
+     */
+    @VisibleForTesting
+    public void removeFromWarnDirs(File dir) {
+        if (warnDirs.contains(dir)) {
+            LOG.warn(dir + " usages is under warn threshold. Remove from warn dirs list");
+            List<File> newDirs = new ArrayList<File>(warnDirs);
+            newDirs.remove(dir);
+            warnDirs = newDirs;
+            for (LedgerDirsListener listener : listeners) {
+                listener.diskUnderWarnThreshold(dir);
+            }
+        }
+    }
+
     /**
      * Returns one of the ledger dir from writable dirs list randomly.
      */
@@ -402,6 +444,12 @@ default void diskFailed(File disk) {}
          */
         default void diskAlmostFull(File disk) {}
 
+        /**
+         * Notified when the disk usage is under warn threshold.
+         * @param disk
+         */
+        default void diskUnderWarnThreshold(File disk) {}
+
         /**
          * This will be notified on disk detected as full.
          *

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerDirsMonitor.java

@@ -91,9 +91,7 @@ private void check(final LedgerDirsManager ldm) {
                         }
                         return e.getUsage();
                     });
-                    for (LedgerDirsListener listener : ldm.getListeners()) {
-                        listener.diskAlmostFull(dir);
-                    }
+                    ldm.addToWarnDirs(dir);
                 } catch (DiskOutOfSpaceException e) {
                     diskUsages.compute(dir, (d, prevUsage) -> {
                         if (null == prevUsage || e.getUsage() != prevUsage) {
@@ -124,6 +122,25 @@ private void check(final LedgerDirsManager ldm) {
             }
         }
 
+        // - Update warnThresholdDirs disk usage.
+        List<File> warnDirs = new ArrayList<File>(ldm.getWarnLedgerDirs());
+        for (File dir : warnDirs) {
+            try {
+                diskUsages.put(dir, diskChecker.checkDir(dir));
+                ldm.removeFromWarnDirs(dir);
+            } catch (DiskErrorException e) {
+                // Notify disk failure to all the listeners
+                for (LedgerDirsListener listener : ldm.getListeners()) {
+                    listener.diskFailed(dir);
+                }
+            } catch (DiskWarnThresholdException e) {
+                diskUsages.put(dir, e.getUsage());
+            } catch (DiskOutOfSpaceException e) {
+                diskUsages.put(dir, e.getUsage());
+                ldm.addToFilledDirs(dir);
+            }
+        }
+
         List<File> fullfilledDirs = new ArrayList<File>(ldm.getFullFilledLedgerDirs());
         boolean makeWritable = ldm.hasWritableLedgerDirs();
 

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java

@@ -1154,6 +1154,15 @@ public void diskAlmostFull(File disk) {
                 }
             }
 
+            @Override
+            public void diskUnderWarnThreshold(File disk) {
+                if (gcThread.isForceGCAllowWhenNoSpace()) {
+                    gcThread.disableForceGC();
+                } else {
+                    gcThread.resumeMajorGC();
+                }
+            }
+
             @Override
             public void diskFull(File disk) {
                 if (gcThread.isForceGCAllowWhenNoSpace()) {

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/LedgerDirsManagerTest.java

@@ -36,6 +36,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import org.apache.bookkeeper.bookie.LedgerDirsManager.LedgerDirsListener;
@@ -280,6 +281,22 @@ public void testAnyLedgerFullTransitToReadOnly(boolean isReadOnlyModeOnAnyDiskFu
         dirsManager.addLedgerDirsListener(mockLedgerDirsListener);
         ledgerMonitor.start();
 
+        final CountDownLatch diskAlmostFull = new CountDownLatch(1);
+        final CountDownLatch diskUnderWarnThreshold = new CountDownLatch(1);
+        dirsManager.addLedgerDirsListener(new LedgerDirsListener() {
+
+            @Override
+            public void diskAlmostFull(File disk) {
+                diskAlmostFull.countDown();
+            }
+
+            @Override
+            public void diskUnderWarnThreshold(File disk) {
+                diskUnderWarnThreshold.countDown();
+            }
+
+        });
+
         Thread.sleep((diskCheckInterval * 2) + 100);
         assertFalse(mockLedgerDirsListener.readOnly);
 
@@ -294,6 +311,14 @@ public void testAnyLedgerFullTransitToReadOnly(boolean isReadOnlyModeOnAnyDiskFu
                 mockLedgerDirsListener, true);
             setUsageAndThenVerify(curDir1, nospace - 0.20f, curDir2, nospace - 0.20f, mockDiskChecker,
                 mockLedgerDirsListener, false);
+            assertTrue("Disk Almost Full should have been  triggered", diskAlmostFull.getCount() == 0);
+            // curDirs1's usage was above warn threshold 0.5 before, now it is below warn threshold 0.5.
+            setUsageAndThenVerify(curDir1, nospace - 0.50f, curDir2,
+                    nospace - 0.20f, mockDiskChecker, mockLedgerDirsListener, false);
+            assertTrue("Disk diskUnderWarnThreshold event should have been triggered",
+                    diskUnderWarnThreshold.getCount() == 0);
+            assertFalse(dirsManager.getWarnLedgerDirs().contains(curDir1));
+            assertTrue(dirsManager.getWarnLedgerDirs().contains(curDir2));
         } else {
             setUsageAndThenVerify(curDir1, 0.1f, curDir2, 0.1f, mockDiskChecker,
                 mockLedgerDirsListener, false);