apache · gaozhangmin · Feb 13, 2023
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java
@@ -288,15 +288,22 @@ public void enableForceGC() {
         if (forceGarbageCollection.compareAndSet(false, true)) {
             LOG.info("Forced garbage collection triggered by thread: {}", Thread.currentThread().getName());
             triggerGC(true, suspendMajorCompaction.get(),
-                      suspendMinorCompaction.get());
+                      suspendMinorCompaction.get(), majorCompactionThreshold, minorCompactionThreshold,
+                    majorCompactionMaxTimeMillis, minorCompactionMaxTimeMillis);
         }
     }
 
-    public void enableForceGC(boolean forceMajor, boolean forceMinor) {
+    public void enableForceGC(boolean forceMajor, boolean forceMinor,
+                              double majorCompactionThreshold, double minorCompactionThreshold,
+                              long majorCompactionMaxTimeMillis, long minorCompactionMaxTimeMillis) {
         if (forceGarbageCollection.compareAndSet(false, true)) {
-            LOG.info("Forced garbage collection triggered by thread: {}, forceMajor: {}, forceMinor: {}",
-                Thread.currentThread().getName(), forceMajor, forceMinor);
-            triggerGC(true, !forceMajor, !forceMinor);
+            LOG.info("Forced garbage collection triggered by thread: {}, forceMajor: {}, forceMinor: {}, "
+                            + "majorCompactionThreshold :{}, minorCompactionThreshold: {}, "
+                            + "majorCompactionMaxTimeMillis: {}, minorCompactionMaxTimeMillis: {}",
+                Thread.currentThread().getName(), forceMajor, forceMinor, majorCompactionThreshold,
+                    minorCompactionThreshold, majorCompactionMaxTimeMillis, minorCompactionMaxTimeMillis);
+            triggerGC(true, !forceMajor, !forceMinor, majorCompactionThreshold, minorCompactionThreshold,
+                    majorCompactionMaxTimeMillis, minorCompactionMaxTimeMillis);
         }
     }
 
@@ -309,9 +316,14 @@ public void disableForceGC() {
 
     Future<?> triggerGC(final boolean force,
                         final boolean suspendMajor,
-                        final boolean suspendMinor) {
+                        final boolean suspendMinor,
+                        final double majorCompactionThreshold,
+                        final double minorCompactionThreshold,
+                        final long majorCompactionMaxTimeMillis,
+                        final long minorCompactionMaxTimeMillis) {
         return gcExecutor.submit(() -> {
-                runWithFlags(force, suspendMajor, suspendMinor);
+                runWithFlags(force, suspendMajor, suspendMinor, majorCompactionThreshold, minorCompactionThreshold,
+                        majorCompactionMaxTimeMillis, minorCompactionMaxTimeMillis);
             });
     }
 
@@ -321,7 +333,8 @@ Future<?> triggerGC() {
         final boolean suspendMinor = suspendMinorCompaction.get();
 
         return gcExecutor.submit(() -> {
-                runWithFlags(force, suspendMajor, suspendMinor);
+                runWithFlags(force, suspendMajor, suspendMinor, majorCompactionThreshold, minorCompactionThreshold,
+                        majorCompactionMaxTimeMillis, minorCompactionMaxTimeMillis);
             });
     }
 
@@ -391,7 +404,8 @@ public void run() {
         boolean suspendMajor = suspendMajorCompaction.get();
         boolean suspendMinor = suspendMinorCompaction.get();
 
-        runWithFlags(force, suspendMajor, suspendMinor);
+        runWithFlags(force, suspendMajor, suspendMinor, majorCompactionThreshold, minorCompactionThreshold,
+                majorCompactionMaxTimeMillis, minorCompactionMaxTimeMillis);
 
         if (force) {
             // only set force to false if it had been true when the garbage
@@ -400,7 +414,9 @@ public void run() {
         }
     }
 
-    public void runWithFlags(boolean force, boolean suspendMajor, boolean suspendMinor) {
+    public void runWithFlags(boolean force, boolean suspendMajor, boolean suspendMinor,
+                             double majorCompactionThreshold, double minorCompactionThreshold,
+                             long majorCompactionMaxTimeMillis, long minorCompactionMaxTimeMillis) {
         long threadStart = MathUtils.nowInNano();
         if (force) {
             LOG.info("Garbage collector thread forced to perform GC before expiry of wait time.");

diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorage.java
@@ -260,8 +260,11 @@ public void forceGC() {
     }
 
     @Override
-    public void forceGC(boolean forceMajor, boolean forceMinor) {
-        gcThread.enableForceGC(forceMajor, forceMinor);
+    public void forceGC(boolean forceMajor, boolean forceMinor,
+                        double majorCompactionThreshold, double minorCompactionThreshold,
+                        long majorCompactionMaxTimeMillis, long minorCompactionMaxTimeMillis) {
+        gcThread.enableForceGC(forceMajor, forceMinor, majorCompactionThreshold, minorCompactionThreshold,
+                majorCompactionMaxTimeMillis, minorCompactionMaxTimeMillis);
     }
 
     @Override

diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerStorage.java
@@ -232,7 +232,9 @@ default void forceGC() {
     /**
      * Force trigger Garbage Collection with forceMajor or forceMinor parameter.
      */
-    default void forceGC(boolean forceMajor, boolean forceMinor) {
+    default void forceGC(boolean forceMajor, boolean forceMinor,
+                         double majorCompactionThreshold, double minorCompactionThreshold,
+                         long majorCompactionMaxTimeMillis, long minorCompactionMaxTimeMillis) {
         return;
     }
 

diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java
@@ -367,8 +367,11 @@ public void forceGC() {
     }
 
     @Override
-    public void forceGC(boolean forceMajor, boolean forceMinor) {
-        interleavedLedgerStorage.forceGC(forceMajor, forceMinor);
+    public void forceGC(boolean forceMajor, boolean forceMinor,
+                        double majorCompactionThreshold, double minorCompactionThreshold,
+                        long majorCompactionMaxTimeMillis, long minorCompactionMaxTimeMillis) {
+        interleavedLedgerStorage.forceGC(forceMajor, forceMinor, majorCompactionThreshold, minorCompactionThreshold,
+                majorCompactionMaxTimeMillis, minorCompactionMaxTimeMillis);
     }
 
     @Override

diff --git a/...keeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorage.java b/...keeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorage.java
@@ -511,8 +511,12 @@ public void forceGC() {
     }
 
     @Override
-    public void forceGC(boolean forceMajor, boolean forceMinor) {
-        ledgerStorageList.stream().forEach(s -> s.forceGC(forceMajor, forceMinor));
+    public void forceGC(boolean forceMajor, boolean forceMinor,
+                        double majorCompactionThreshold, double minorCompactionThreshold,
+                        long majorCompactionMaxTimeMillis, long minorCompactionMaxTimeMillis) {
+        ledgerStorageList.stream().forEach(s -> s.forceGC(forceMajor, forceMinor,
+                majorCompactionThreshold, minorCompactionThreshold,
+                majorCompactionMaxTimeMillis, minorCompactionMaxTimeMillis));
     }
 
     @Override

diff --git a/...rc/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java b/...rc/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java
@@ -272,8 +272,11 @@ public void forceGC() {
     }
 
     @Override
-    public void forceGC(boolean forceMajor, boolean forceMinor) {
-        gcThread.enableForceGC(forceMajor, forceMinor);
+    public void forceGC(boolean forceMajor, boolean forceMinor,
+                        double majorCompactionThreshold, double minorCompactionThreshold,
+                        long majorCompactionMaxTimeMillis, long minorCompactionMaxTimeMillis) {
+        gcThread.enableForceGC(forceMajor, forceMinor, majorCompactionThreshold, minorCompactionThreshold,
+                majorCompactionMaxTimeMillis, minorCompactionMaxTimeMillis);
     }
 
     @Override

diff --git a/...eper-server/src/main/java/org/apache/bookkeeper/server/http/service/TriggerGCService.java b/...eper-server/src/main/java/org/apache/bookkeeper/server/http/service/TriggerGCService.java
@@ -77,7 +77,29 @@ public HttpServiceResponse handle(HttpServiceRequest request) throws Exception {
 
                     forceMajor = Boolean.parseBoolean(configMap.getOrDefault("forceMajor", forceMajor).toString());
                     forceMinor = Boolean.parseBoolean(configMap.getOrDefault("forceMinor", forceMinor).toString());
-                    ledgerStorage.forceGC(forceMajor, forceMinor);
+
+                    double majorCompactionThreshold = (double) configMap.getOrDefault("majorCompactionThreshold",
+                            conf.getMajorCompactionThreshold());
+                    double minorCompactionThreshold = (double) configMap.getOrDefault("minorCompactionThreshold",
+                            conf.getMinorCompactionThreshold());
+                    long majorCompactionMaxTimeMillis =  configMap.get(
+                            "majorCompactionMaxTimeMillis") == null ? conf.getMajorCompactionMaxTimeMillis()
+                            : (long) (int) configMap.get("majorCompactionMaxTimeMillis");
+                    long minorCompactionMaxTimeMillis =  configMap.get(
+                            "minorCompactionMaxTimeMillis") == null ? conf.getMinorCompactionMaxTimeMillis()
+                            : (long) (int) configMap.get("majorCompactionMaxTimeMillis");
+
+                    if (majorCompactionThreshold > 1.0f || majorCompactionThreshold < 0
+                            || minorCompactionThreshold > 1.0f || minorCompactionThreshold < 0
+                            || minorCompactionThreshold >= majorCompactionThreshold) {
+                        response.setCode(HttpServer.StatusCode.BAD_REQUEST);
+                        response.setBody("Bad request parameters");
+                        return response;
+                    }
+
+                    ledgerStorage.forceGC(forceMajor, forceMinor,
+                            majorCompactionThreshold, minorCompactionThreshold,
+                            majorCompactionMaxTimeMillis, minorCompactionMaxTimeMillis);
                 }
 
                 String output = "Triggered GC on BookieServer: " + bookieServer.getBookieId();

diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionTest.java
@@ -362,7 +362,8 @@ public void testForceGarbageCollectionWhenDiskIsFull(boolean isForceCompactionAl
         // disable forceMajor and forceMinor
         majorCompactionCntBeforeGC = getGCThread().getGarbageCollectionStatus().getMajorCompactionCounter();
         minorCompactionCntBeforeGC = getGCThread().getGarbageCollectionStatus().getMinorCompactionCounter();
-        getGCThread().triggerGC(true, true, true).get();
+        getGCThread().triggerGC(true, true, true, majorCompactionThreshold, minorCompactionThreshold,
+                baseConf.getMajorCompactionMaxTimeMillis(), baseConf.getMinorCompactionMaxTimeMillis()).get();
         majorCompactionCntAfterGC = getGCThread().getGarbageCollectionStatus().getMajorCompactionCounter();
         minorCompactionCntAfterGC = getGCThread().getGarbageCollectionStatus().getMinorCompactionCounter();
         assertEquals(majorCompactionCntBeforeGC, majorCompactionCntAfterGC);
@@ -371,7 +372,8 @@ public void testForceGarbageCollectionWhenDiskIsFull(boolean isForceCompactionAl
         // enable forceMajor and forceMinor
         majorCompactionCntBeforeGC = getGCThread().getGarbageCollectionStatus().getMajorCompactionCounter();
         minorCompactionCntBeforeGC = getGCThread().getGarbageCollectionStatus().getMinorCompactionCounter();
-        getGCThread().triggerGC(true, false, false).get();
+        getGCThread().triggerGC(true, false, false, majorCompactionThreshold, minorCompactionThreshold,
+                baseConf.getMajorCompactionMaxTimeMillis(), baseConf.getMinorCompactionMaxTimeMillis()).get();
         majorCompactionCntAfterGC = getGCThread().getGarbageCollectionStatus().getMajorCompactionCounter();
         minorCompactionCntAfterGC = getGCThread().getGarbageCollectionStatus().getMinorCompactionCounter();
         assertEquals(majorCompactionCntBeforeGC + 1, majorCompactionCntAfterGC);
@@ -380,7 +382,8 @@ public void testForceGarbageCollectionWhenDiskIsFull(boolean isForceCompactionAl
         // enable forceMajor and disable forceMinor
         majorCompactionCntBeforeGC = getGCThread().getGarbageCollectionStatus().getMajorCompactionCounter();
         minorCompactionCntBeforeGC = getGCThread().getGarbageCollectionStatus().getMinorCompactionCounter();
-        getGCThread().triggerGC(true, false, true).get();
+        getGCThread().triggerGC(true, false, true, majorCompactionThreshold, minorCompactionThreshold,
+                baseConf.getMajorCompactionMaxTimeMillis(), baseConf.getMinorCompactionMaxTimeMillis()).get();
         majorCompactionCntAfterGC = getGCThread().getGarbageCollectionStatus().getMajorCompactionCounter();
         minorCompactionCntAfterGC = getGCThread().getGarbageCollectionStatus().getMinorCompactionCounter();
         assertEquals(majorCompactionCntBeforeGC + 1, majorCompactionCntAfterGC);
@@ -389,7 +392,8 @@ public void testForceGarbageCollectionWhenDiskIsFull(boolean isForceCompactionAl
         // disable forceMajor and enable forceMinor
         majorCompactionCntBeforeGC = getGCThread().getGarbageCollectionStatus().getMajorCompactionCounter();
         minorCompactionCntBeforeGC = getGCThread().getGarbageCollectionStatus().getMinorCompactionCounter();
-        getGCThread().triggerGC(true, true, false).get();
+        getGCThread().triggerGC(true, true, false, majorCompactionThreshold, minorCompactionThreshold,
+                baseConf.getMajorCompactionMaxTimeMillis(), baseConf.getMinorCompactionMaxTimeMillis()).get();
         majorCompactionCntAfterGC = getGCThread().getGarbageCollectionStatus().getMajorCompactionCounter();
         minorCompactionCntAfterGC = getGCThread().getGarbageCollectionStatus().getMinorCompactionCounter();
         assertEquals(majorCompactionCntBeforeGC, majorCompactionCntAfterGC);
@@ -721,7 +725,8 @@ public void testMinorCompactionWithEntryLogPerLedgerEnabled() throws Exception {
         }
 
         LOG.info("Finished deleting the ledgers contains most entries.");
-        getGCThread().triggerGC(true, false, false).get();
+        getGCThread().triggerGC(true, false, false, majorCompactionThreshold, minorCompactionThreshold,
+                baseConf.getMajorCompactionMaxTimeMillis(), baseConf.getMinorCompactionMaxTimeMillis()).get();
 
         assertEquals(lastMajorCompactionTime, getGCThread().lastMajorCompactionTime);
         assertTrue(getGCThread().lastMinorCompactionTime > lastMinorCompactionTime);
@@ -740,17 +745,20 @@ public void testMinorCompactionWithEntryLogPerLedgerEnabled() throws Exception {
         // Now, let's mark E1 as flushed, as its ledger L1 has been deleted already. In this case, the GC algorithm
         // should consider it for deletion.
         ((DefaultEntryLogger) getGCThread().entryLogger).recentlyCreatedEntryLogsStatus.flushRotatedEntryLog(1L);
-        getGCThread().triggerGC(true, false, false).get();
+        getGCThread().triggerGC(true, false, false, majorCompactionThreshold, minorCompactionThreshold,
+                baseConf.getMajorCompactionMaxTimeMillis(), baseConf.getMinorCompactionMaxTimeMillis()).get();
         assertTrue("Found entry log file 1.log that should have been compacted in ledgerDirectory: "
                 + tmpDirs.getDirs().get(0), TestUtils.hasNoneLogFiles(tmpDirs.getDirs().get(0), 1));
 
         // Once removed the ledger L0, then deleting E0 is fine (only if it has been flushed).
         bkc.deleteLedger(lhs[0].getId());
-        getGCThread().triggerGC(true, false, false).get();
+        getGCThread().triggerGC(true, false, false, majorCompactionThreshold, minorCompactionThreshold,
+                baseConf.getMajorCompactionMaxTimeMillis(), baseConf.getMinorCompactionMaxTimeMillis()).get();
         assertTrue("Found entry log file 0.log that should not have been compacted in ledgerDirectory: "
                 + tmpDirs.getDirs().get(0), TestUtils.hasAllLogFiles(tmpDirs.getDirs().get(0), 0));
         ((DefaultEntryLogger) getGCThread().entryLogger).recentlyCreatedEntryLogsStatus.flushRotatedEntryLog(0L);
-        getGCThread().triggerGC(true, false, false).get();
+        getGCThread().triggerGC(true, false, false, majorCompactionThreshold, minorCompactionThreshold,
+                baseConf.getMajorCompactionMaxTimeMillis(), baseConf.getMinorCompactionMaxTimeMillis()).get();
         assertTrue("Found entry log file 0.log that should have been compacted in ledgerDirectory: "
                 + tmpDirs.getDirs().get(0), TestUtils.hasNoneLogFiles(tmpDirs.getDirs().get(0), 0));
     }
@@ -868,7 +876,8 @@ public void testMinorCompactionWithNoWritableLedgerDirsButIsForceGCAllowWhenNoSp
         bkc.deleteLedger(lhs[2].getId());
 
         LOG.info("Finished deleting the ledgers contains most entries.");
-        getGCThread().triggerGC(true, false, false).get();
+        getGCThread().triggerGC(true, false, false, majorCompactionThreshold, minorCompactionThreshold,
+                baseConf.getMajorCompactionMaxTimeMillis(), baseConf.getMinorCompactionMaxTimeMillis()).get();
 
         // after garbage collection, major compaction should not be executed
         assertEquals(lastMajorCompactionTime, getGCThread().lastMajorCompactionTime);
@@ -1407,7 +1416,8 @@ public void testCancelledCompactionWhenShuttingDown() throws Exception {
         bkc.deleteLedger(lhs[2].getId());
         LOG.info("Finished deleting the ledgers contains most entries.");
 
-        getGCThread().triggerGC(true, false, false);
+        getGCThread().triggerGC(true, false, false, majorCompactionThreshold, minorCompactionThreshold,
+                baseConf.getMajorCompactionMaxTimeMillis(), baseConf.getMinorCompactionMaxTimeMillis());
         getGCThread().throttler.cancelledAcquire();
         waitUntilTrue(() -> {
             try {

diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GarbageCollectorThreadTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GarbageCollectorThreadTest.java
@@ -295,7 +295,9 @@ public void testCompactionWithFileSizeCheck() throws Exception {
 
         storage.deleteLedger(1);
         // only logId 1 will be compacted.
-        gcThread.runWithFlags(true, true, false);
+        gcThread.runWithFlags(true, true, false, conf.getMajorCompactionThreshold(),
+                conf.getMinorCompactionThreshold(), conf.getMajorCompactionMaxTimeMillis(),
+                conf.getMinorCompactionMaxTimeMillis());
 
         // logId1 and logId2 should be compacted
         assertFalse(entryLogger.logExists(logId1));
@@ -362,7 +364,9 @@ public void testCompactionWithoutFileSizeCheck() throws Exception {
         assertTrue(entryLogMetaMap.containsKey(logId2));
         assertTrue(entryLogger.logExists(logId3));
 
-        gcThread.runWithFlags(true, true, false);
+        gcThread.runWithFlags(true, true, false, conf.getMajorCompactionThreshold(),
+                conf.getMinorCompactionThreshold(), conf.getMajorCompactionMaxTimeMillis(),
+                conf.getMinorCompactionMaxTimeMillis());
 
         assertTrue(entryLogger.logExists(logId1));
         assertTrue(entryLogger.logExists(logId2));

diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/MockLedgerStorage.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/MockLedgerStorage.java
@@ -271,8 +271,12 @@ public void forceGC() {
     }
 
     @Override
-    public void forceGC(boolean forceMajor, boolean forceMinor) {
-        CompactableLedgerStorage.super.forceGC(forceMajor, forceMinor);
+    public void forceGC(boolean forceMajor, boolean forceMinor,
+                        double majorCompactionThreshold, double minorCompactionThreshold,
+                        long majorCompactionMaxTimeMillis, long minorCompactionMaxTimeMillis) {
+        CompactableLedgerStorage.super.forceGC(forceMajor, forceMinor,
+                majorCompactionThreshold, minorCompactionThreshold,
+                majorCompactionMaxTimeMillis, minorCompactionMaxTimeMillis);
     }
 
     public void suspendMinorGC() {