Allow ML offload immediately after ledger closed (apache#1965)

The documentation said that only negative values disabled. If a user
wants data to be offloaded as soon as possible, the obvious thing is
to set the threshold to 0. Previously this disabled as the check
was > 0, rather than >=.

This patch changes the ML implementation to accept 0 as a threshold
and adds a test to ensure if specified, the ledgers are offloaded as
soon as possible.

Master Issue: apache#1511

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java

@@ -412,7 +412,7 @@ public long getOffloadLedgerDeletionLagMillis() {
 
     /**
      * Size, in bytes, at which the managed ledger will start to automatically offload ledgers to longterm storage.
-     * A negative value disables autotriggering.
+     * A negative value disables autotriggering. A threshold of 0 offloads data as soon as possible.
      * Offloading will not occur if no offloader has been set {@link #setLedgerOffloader(LedgerOffloader)}.
      * Automatical offloading occurs when the ledger is rolled, and the ledgers up to that point exceed the threshold.
      *

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

@@ -1582,7 +1582,7 @@ private void scheduleDeferredTrimming(CompletableFuture<?> promise) {
     }
 
     private void maybeOffloadInBackground(CompletableFuture<PositionImpl> promise) {
-        if (config.getOffloadAutoTriggerSizeThresholdBytes() > 0) {
+        if (config.getOffloadAutoTriggerSizeThresholdBytes() >= 0) {
             executor.executeOrdered(name, safeRun(() -> maybeOffload(promise)));
         }
     }

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixTest.java

@@ -865,6 +865,37 @@ public CompletableFuture<Void> offload(ReadHandle ledger,
                                             ledger.getLedgersInfoAsList().get(2).getLedgerId()));
     }
 
+    @Test
+    public void offloadAsSoonAsClosed() throws Exception {
+
+        MockLedgerOffloader offloader = new MockLedgerOffloader();
+        ManagedLedgerConfig config = new ManagedLedgerConfig();
+        config.setMaxEntriesPerLedger(10);
+        config.setOffloadAutoTriggerSizeThresholdBytes(0);
+        config.setRetentionTime(10, TimeUnit.MINUTES);
+        config.setLedgerOffloader(offloader);
+
+        ManagedLedgerImpl ledger = (ManagedLedgerImpl)factory.open("my_test_ledger", config);
+
+        for (int i = 0; i < 11; i++) {
+            ledger.addEntry(buildEntry(10, "entry-" + i));
+        }
+
+        assertEventuallyTrue(() -> offloader.offloadedLedgers().size() == 1);
+        Assert.assertEquals(offloader.offloadedLedgers(),
+                            ImmutableSet.of(ledger.getLedgersInfoAsList().get(0).getLedgerId()));
+
+        for (int i = 0; i < 10; i++) {
+            ledger.addEntry(buildEntry(10, "entry-" + i));
+        }
+
+        assertEventuallyTrue(() -> offloader.offloadedLedgers().size() == 2);
+        Assert.assertEquals(offloader.offloadedLedgers(),
+                            ImmutableSet.of(ledger.getLedgersInfoAsList().get(0).getLedgerId(),
+                                            ledger.getLedgersInfoAsList().get(1).getLedgerId()));
+    }
+
+
     static void assertEventuallyTrue(BooleanSupplier predicate) throws Exception {
         // wait up to 3 seconds
         for (int i = 0; i < 30 && !predicate.getAsBoolean(); i++) {