[fix] [broker] Fix system topic can not be loaded up if it contains d…

…ata offloaded (#23279)

(cherry picked from commit fc0e4e3)

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

@@ -230,5 +230,9 @@ default void scanLedgers(OffloadedLedgerMetadataConsumer consumer,
                              Map<String, String> offloadDriverMetadata) throws ManagedLedgerException {
         throw ManagedLedgerException.getManagedLedgerException(new UnsupportedOperationException());
     }
+
+    default boolean isAppendable() {
+        return true;
+    }
 }
 

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

@@ -94,6 +94,7 @@
 import org.apache.bookkeeper.mledger.AsyncCallbacks.TerminateCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.UpdatePropertiesCallback;
 import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.LedgerOffloader;
 import org.apache.bookkeeper.mledger.ManagedCursor;
 import org.apache.bookkeeper.mledger.ManagedLedger;
 import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
@@ -2484,8 +2485,7 @@ private void scheduleDeferredTrimming(boolean isTruncate, CompletableFuture<?> p
     }
 
     public void maybeOffloadInBackground(CompletableFuture<PositionImpl> promise) {
-        if (config.getLedgerOffloader() == null || config.getLedgerOffloader() == NullLedgerOffloader.INSTANCE
-                || config.getLedgerOffloader().getOffloadPolicies() == null) {
+        if (getOffloadPoliciesIfAppendable().isEmpty()) {
             return;
         }
 
@@ -2501,8 +2501,7 @@ public void maybeOffloadInBackground(CompletableFuture<PositionImpl> promise) {
 
     private void maybeOffload(long offloadThresholdInBytes, long offloadThresholdInSeconds,
                               CompletableFuture<PositionImpl> finalPromise) {
-        if (config.getLedgerOffloader() == null || config.getLedgerOffloader() == NullLedgerOffloader.INSTANCE
-                || config.getLedgerOffloader().getOffloadPolicies() == null) {
+        if (getOffloadPoliciesIfAppendable().isEmpty()) {
             String msg = String.format("[%s] Nothing to offload due to offloader or offloadPolicies is NULL", name);
             finalPromise.completeExceptionally(new IllegalArgumentException(msg));
             return;
@@ -2604,6 +2603,16 @@ void internalTrimConsumedLedgers(CompletableFuture<?> promise) {
         internalTrimLedgers(false, promise);
     }
 
+    private Optional<OffloadPolicies> getOffloadPoliciesIfAppendable() {
+        LedgerOffloader ledgerOffloader = config.getLedgerOffloader();
+        if (ledgerOffloader == null
+                || !ledgerOffloader.isAppendable()
+                || ledgerOffloader.getOffloadPolicies() == null) {
+            return Optional.empty();
+        }
+        return Optional.ofNullable(ledgerOffloader.getOffloadPolicies());
+    }
+
     void internalTrimLedgers(boolean isTruncate, CompletableFuture<?> promise) {
         if (!factory.isMetadataServiceAvailable()) {
             // Defer trimming of ledger if we cannot connect to metadata service
@@ -2619,10 +2628,7 @@ void internalTrimLedgers(boolean isTruncate, CompletableFuture<?> promise) {
 
         List<LedgerInfo> ledgersToDelete = new ArrayList<>();
         List<LedgerInfo> offloadedLedgersToDelete = new ArrayList<>();
-        Optional<OffloadPolicies> optionalOffloadPolicies = Optional.ofNullable(config.getLedgerOffloader() != null
-                && config.getLedgerOffloader() != NullLedgerOffloader.INSTANCE
-                ? config.getLedgerOffloader().getOffloadPolicies()
-                : null);
+        Optional<OffloadPolicies> optionalOffloadPolicies = getOffloadPoliciesIfAppendable();
         synchronized (this) {
             if (log.isDebugEnabled()) {
                 log.debug("[{}] Start TrimConsumedLedgers. ledgers={} totalSize={}", name, ledgers.keySet(),
@@ -3141,8 +3147,10 @@ public void offloadFailed(ManagedLedgerException e, Object ctx) {
 
     @Override
     public void asyncOffloadPrefix(Position pos, OffloadCallback callback, Object ctx) {
-        if (config.getLedgerOffloader() != null && config.getLedgerOffloader() == NullLedgerOffloader.INSTANCE) {
-            callback.offloadFailed(new ManagedLedgerException("NullLedgerOffloader"), ctx);
+        LedgerOffloader ledgerOffloader = config.getLedgerOffloader();
+        if (ledgerOffloader != null && !ledgerOffloader.isAppendable()) {
+            String msg = String.format("[%s] does not support offload", ledgerOffloader.getClass().getSimpleName());
+            callback.offloadFailed(new ManagedLedgerException(msg), ctx);
             return;
         }
         PositionImpl requestOffloadTo = (PositionImpl) pos;

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

@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.mledger.impl;
+
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.mledger.LedgerOffloader;
+import org.apache.pulsar.common.policies.data.OffloadPolicies;
+import org.apache.pulsar.common.util.FutureUtil;
+
+public class NonAppendableLedgerOffloader implements LedgerOffloader {
+    private LedgerOffloader delegate;
+
+    public NonAppendableLedgerOffloader(LedgerOffloader delegate) {
+        this.delegate = delegate;
+    }
+
+    @Override
+    public String getOffloadDriverName() {
+        return delegate.getOffloadDriverName();
+    }
+
+    @Override
+    public CompletableFuture<Void> offload(ReadHandle ledger,
+                                           UUID uid,
+                                           Map<String, String> extraMetadata) {
+        return FutureUtil.failedFuture(new UnsupportedOperationException());
+    }
+
+    @Override
+    public CompletableFuture<ReadHandle> readOffloaded(long ledgerId, UUID uid,
+                                                       Map<String, String> offloadDriverMetadata) {
+        return delegate.readOffloaded(ledgerId, uid, offloadDriverMetadata);
+    }
+
+    @Override
+    public CompletableFuture<Void> deleteOffloaded(long ledgerId, UUID uid,
+                                                   Map<String, String> offloadDriverMetadata) {
+        return delegate.deleteOffloaded(ledgerId, uid, offloadDriverMetadata);
+    }
+
+    @Override
+    public OffloadPolicies getOffloadPolicies() {
+        return delegate.getOffloadPolicies();
+    }
+
+    @Override
+    public void close() {
+        delegate.close();
+    }
+
+    @Override
+    public boolean isAppendable() {
+        return false;
+    }
+}

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

@@ -70,4 +70,9 @@ public OffloadPolicies getOffloadPolicies() {
     public void close() {
 
     }
+
+    @Override
+    public boolean isAppendable() {
+        return false;
+    }
 }

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

@@ -3848,7 +3848,7 @@ public void testDoNotGetOffloadPoliciesMultipleTimesWhenTrimLedgers() throws Exc
         config.setLedgerOffloader(ledgerOffloader);
 
         ledger.internalTrimConsumedLedgers(Futures.NULL_PROMISE);
-        verify(ledgerOffloader, times(1)).getOffloadPolicies();
+        verify(ledgerOffloader, times(1)).isAppendable();
     }
 
     @Test(timeOut = 30000)

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

@@ -29,6 +29,7 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
 import io.netty.buffer.ByteBuf;
 
 import java.util.ArrayList;
@@ -54,18 +55,42 @@
 import org.apache.bookkeeper.mledger.LedgerOffloader;
 import org.apache.bookkeeper.mledger.ManagedCursor;
 import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig;
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo;
 import org.apache.bookkeeper.mledger.util.MockClock;
 import org.apache.bookkeeper.net.BookieId;
 import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
 import org.apache.pulsar.common.policies.data.OffloadPoliciesImpl;
 import org.apache.pulsar.common.policies.data.OffloadedReadPriority;
+import org.awaitility.Awaitility;
 import org.testng.Assert;
+import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
 public class OffloadPrefixReadTest extends MockedBookKeeperTestCase {
-    @Test
-    public void testOffloadRead() throws Exception {
+
+    private final String offloadTypeAppendable = "NonAppendable";
+
+    @Override
+    protected void initManagedLedgerFactoryConfig(ManagedLedgerFactoryConfig config) {
+        super.initManagedLedgerFactoryConfig(config);
+        // disable cache.
+        config.setMaxCacheSize(0);
+    }
+
+    @DataProvider(name = "offloadAndDeleteTypes")
+    public Object[][] offloadAndDeleteTypes() {
+        return new Object[][]{
+                {"normal", true},
+                {"normal", false},
+                {offloadTypeAppendable, true},
+                {offloadTypeAppendable, false},
+        };
+    }
+
+    @Test(dataProvider = "offloadAndDeleteTypes")
+    public void testOffloadRead(String offloadType, boolean deleteMl) throws Exception {
         MockLedgerOffloader offloader = spy(MockLedgerOffloader.class);
         ManagedLedgerConfig config = new ManagedLedgerConfig();
         config.setMaxEntriesPerLedger(10);
@@ -88,6 +113,10 @@ public void testOffloadRead() throws Exception {
         Assert.assertTrue(ledger.getLedgersInfoAsList().get(1).getOffloadContext().getComplete());
         Assert.assertFalse(ledger.getLedgersInfoAsList().get(2).getOffloadContext().getComplete());
 
+        if (offloadTypeAppendable.equals(offloadType)) {
+            config.setLedgerOffloader(new NonAppendableLedgerOffloader(offloader));
+        }
+
         UUID firstLedgerUUID = new UUID(ledger.getLedgersInfoAsList().get(0).getOffloadContext().getUidMsb(),
                 ledger.getLedgersInfoAsList().get(0).getOffloadContext().getUidLsb());
         UUID secondLedgerUUID = new UUID(ledger.getLedgersInfoAsList().get(1).getOffloadContext().getUidMsb(),
@@ -115,13 +144,30 @@ public void testOffloadRead() throws Exception {
         verify(offloader, times(2))
                 .readOffloaded(anyLong(), (UUID) any(), anyMap());
 
-        ledger.close();
-        // Ensure that all the read handles had been closed
-        assertEquals(offloader.openedReadHandles.get(), 0);
+        if (!deleteMl) {
+            ledger.close();
+            // Ensure that all the read handles had been closed
+            assertEquals(offloader.openedReadHandles.get(), 0);
+        } else {
+            // Verify: the ledger offloaded will be deleted after managed ledger is deleted.
+            ledger.delete();
+            Awaitility.await().untilAsserted(() -> {
+                assertTrue(offloader.offloads.size() <= 1);
+                assertTrue(ledger.ledgers.size() <= 1);
+            });
+        }
     }
 
-    @Test
-    public void testBookkeeperFirstOffloadRead() throws Exception {
+    @DataProvider(name = "offloadTypes")
+    public Object[][] offloadTypes() {
+        return new Object[][]{
+                {"normal"},
+                {offloadTypeAppendable},
+        };
+    }
+
+    @Test(dataProvider = "offloadTypes")
+    public void testBookkeeperFirstOffloadRead(String offloadType) throws Exception {
         MockLedgerOffloader offloader = spy(MockLedgerOffloader.class);
         MockClock clock = new MockClock();
         offloader.getOffloadPolicies()
@@ -186,6 +232,10 @@ public void testBookkeeperFirstOffloadRead() throws Exception {
         Assert.assertTrue(ledger.getLedgersInfoAsList().get(0).getOffloadContext().getBookkeeperDeleted());
         Assert.assertTrue(ledger.getLedgersInfoAsList().get(1).getOffloadContext().getBookkeeperDeleted());
 
+        if (offloadTypeAppendable.equals(offloadType)) {
+            config.setLedgerOffloader(new NonAppendableLedgerOffloader(offloader));
+        }
+
         for (Entry e : cursor.readEntries(10)) {
             Assert.assertEquals(new String(e.getData()), "entry-" + i++);
         }
@@ -195,6 +245,56 @@ public void testBookkeeperFirstOffloadRead() throws Exception {
                 .readOffloaded(anyLong(), (UUID) any(), anyMap());
         verify(offloader).readOffloaded(anyLong(), eq(secondLedgerUUID), anyMap());
 
+        // Verify: the ledger offloaded will be trimmed after if no backlog.
+        while (cursor.hasMoreEntries()) {
+            cursor.readEntries(1);
+        }
+        config.setRetentionTime(0, TimeUnit.MILLISECONDS);
+        config.setRetentionSizeInMB(0);
+        CompletableFuture trimFuture = new CompletableFuture();
+        ledger.trimConsumedLedgersInBackground(trimFuture);
+        trimFuture.join();
+        Awaitility.await().untilAsserted(() -> {
+            assertTrue(offloader.offloads.size() <= 1);
+            assertTrue(ledger.ledgers.size() <= 1);
+        });
+
+        // cleanup.
+        ledger.delete();
+    }
+
+
+
+    @Test
+    public void testSkipOffloadIfReadOnly() throws Exception {
+        LedgerOffloader ol = new NonAppendableLedgerOffloader(spy(MockLedgerOffloader.class));
+        ManagedLedgerConfig config = new ManagedLedgerConfig();
+        config.setMaxEntriesPerLedger(10);
+        config.setMinimumRolloverTime(0, TimeUnit.SECONDS);
+        config.setRetentionTime(10, TimeUnit.MINUTES);
+        config.setRetentionSizeInMB(10);
+        config.setLedgerOffloader(ol);
+        ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("my_test_ledger", config);
+
+        for (int i = 0; i < 25; i++) {
+            String content = "entry-" + i;
+            ledger.addEntry(content.getBytes());
+        }
+        assertEquals(ledger.getLedgersInfoAsList().size(), 3);
+
+        try {
+            ledger.offloadPrefix(ledger.getLastConfirmedEntry());
+        } catch (ManagedLedgerException mle) {
+            assertTrue(mle.getMessage().contains("does not support offload"));
+        }
+
+        assertEquals(ledger.getLedgersInfoAsList().size(), 3);
+        Assert.assertFalse(ledger.getLedgersInfoAsList().get(0).getOffloadContext().getComplete());
+        Assert.assertFalse(ledger.getLedgersInfoAsList().get(1).getOffloadContext().getComplete());
+        Assert.assertFalse(ledger.getLedgersInfoAsList().get(2).getOffloadContext().getComplete());
+
+        // cleanup.
+        ledger.delete();
     }
 
 

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

@@ -94,7 +94,7 @@ public void testNullOffloader() throws Exception {
             ledger.offloadPrefix(p);
             fail("Should have thrown an exception");
         } catch (ManagedLedgerException e) {
-            assertEquals(e.getMessage(), "NullLedgerOffloader");
+            assertTrue(e.getMessage().contains("does not support offload"));
         }
         assertEquals(ledger.getLedgersInfoAsList().size(), 5);
         assertEquals(ledger.getLedgersInfoAsList().stream()

managed-ledger/src/test/java/org/apache/bookkeeper/test/MockedBookKeeperTestCase.java

@@ -83,13 +83,17 @@ public final void setUp(Method method) throws Exception {
         }
 
         ManagedLedgerFactoryConfig managedLedgerFactoryConfig = new ManagedLedgerFactoryConfig();
-        // increase default cache eviction interval so that caching could be tested with less flakyness
-        managedLedgerFactoryConfig.setCacheEvictionIntervalMs(200);
+        initManagedLedgerFactoryConfig(managedLedgerFactoryConfig);
         factory = new ManagedLedgerFactoryImpl(metadataStore, bkc);
 
         setUpTestCase();
     }
 
+    protected void initManagedLedgerFactoryConfig(ManagedLedgerFactoryConfig config) {
+        // increase default cache eviction interval so that caching could be tested with less flakyness
+        config.setCacheEvictionIntervalMs(200);
+    }
+
     protected void setUpTestCase() throws Exception {
 
     }