From e4d69483f9a3e6efc4dbabe4dda31dde4e03639f Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Wed, 11 Sep 2024 19:33:47 +0800 Subject: [PATCH] [fix] [broker] Fix system topic can not be loaded up if it contains data offloaded (#23279) --- .../bookkeeper/mledger/LedgerOffloader.java | 4 + .../mledger/impl/ManagedLedgerImpl.java | 28 +++-- .../impl/NonAppendableLedgerOffloader.java | 74 ++++++++++++ .../mledger/impl/NullLedgerOffloader.java | 5 + .../mledger/impl/ManagedLedgerTest.java | 2 +- .../mledger/impl/OffloadPrefixReadTest.java | 114 ++++++++++++++++-- .../mledger/impl/OffloadPrefixTest.java | 2 +- .../test/MockedBookKeeperTestCase.java | 8 +- .../pulsar/broker/service/BrokerService.java | 41 +++---- .../broker/admin/AdminApiOffloadTest.java | 1 + .../broker/service/BrokerServiceTest.java | 21 +++- 11 files changed, 255 insertions(+), 45 deletions(-) create mode 100644 managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonAppendableLedgerOffloader.java diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/LedgerOffloader.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/LedgerOffloader.java index 868a8e4265365c..11148ef1a59f50 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/LedgerOffloader.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/LedgerOffloader.java @@ -230,5 +230,9 @@ default void scanLedgers(OffloadedLedgerMetadataConsumer consumer, Map offloadDriverMetadata) throws ManagedLedgerException { throw ManagedLedgerException.getManagedLedgerException(new UnsupportedOperationException()); } + + default boolean isAppendable() { + return true; + } } diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java index 92c55e572a49a6..8cb5a3ee6aceca 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java +++ b/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.ManagedLedgerAttributes; @@ -2451,8 +2452,7 @@ private void scheduleDeferredTrimming(boolean isTruncate, CompletableFuture p } public void maybeOffloadInBackground(CompletableFuture promise) { - if (config.getLedgerOffloader() == null || config.getLedgerOffloader() == NullLedgerOffloader.INSTANCE - || config.getLedgerOffloader().getOffloadPolicies() == null) { + if (getOffloadPoliciesIfAppendable().isEmpty()) { return; } @@ -2468,8 +2468,7 @@ public void maybeOffloadInBackground(CompletableFuture promise) { private void maybeOffload(long offloadThresholdInBytes, long offloadThresholdInSeconds, CompletableFuture 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; @@ -2572,6 +2571,16 @@ void internalTrimConsumedLedgers(CompletableFuture promise) { internalTrimLedgers(false, promise); } + private Optional 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 @@ -2587,10 +2596,7 @@ void internalTrimLedgers(boolean isTruncate, CompletableFuture promise) { List ledgersToDelete = new ArrayList<>(); List offloadedLedgersToDelete = new ArrayList<>(); - Optional optionalOffloadPolicies = Optional.ofNullable(config.getLedgerOffloader() != null - && config.getLedgerOffloader() != NullLedgerOffloader.INSTANCE - ? config.getLedgerOffloader().getOffloadPolicies() - : null); + Optional optionalOffloadPolicies = getOffloadPoliciesIfAppendable(); synchronized (this) { if (log.isDebugEnabled()) { log.debug("[{}] Start TrimConsumedLedgers. ledgers={} totalSize={}", name, ledgers.keySet(), @@ -3117,8 +3123,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; } Position requestOffloadTo = pos; diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonAppendableLedgerOffloader.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonAppendableLedgerOffloader.java new file mode 100644 index 00000000000000..f3001ec8050e21 --- /dev/null +++ b/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 offload(ReadHandle ledger, + UUID uid, + Map extraMetadata) { + return FutureUtil.failedFuture(new UnsupportedOperationException()); + } + + @Override + public CompletableFuture readOffloaded(long ledgerId, UUID uid, + Map offloadDriverMetadata) { + return delegate.readOffloaded(ledgerId, uid, offloadDriverMetadata); + } + + @Override + public CompletableFuture deleteOffloaded(long ledgerId, UUID uid, + Map 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; + } +} diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NullLedgerOffloader.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NullLedgerOffloader.java index 938ceb0c7dfbc1..fe646bc82e55ae 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NullLedgerOffloader.java +++ b/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; + } } diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java index e3b272babb7bba..bb38114ef71176 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java @@ -3849,7 +3849,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) diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixReadTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixReadTest.java index 6d8ecba8688470..48751417e1714c 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixReadTest.java +++ b/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,6 +55,8 @@ 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.PositionFactory; import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo; import org.apache.bookkeeper.mledger.util.MockClock; @@ -61,12 +64,34 @@ 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); @@ -89,6 +114,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(), @@ -116,13 +145,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() @@ -187,6 +233,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++); } @@ -196,6 +246,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(); } diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixTest.java index 331e7b03173944..3f9f4f8da12f26 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixTest.java @@ -95,7 +95,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() diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/test/MockedBookKeeperTestCase.java b/managed-ledger/src/test/java/org/apache/bookkeeper/test/MockedBookKeeperTestCase.java index 645563eb78c4d8..c7685cfaa65944 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/test/MockedBookKeeperTestCase.java +++ b/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 { } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java index c0e3e7d356be0a..17e5288b5f1798 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java @@ -94,7 +94,7 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException; import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerNotFoundException; import org.apache.bookkeeper.mledger.ManagedLedgerFactory; -import org.apache.bookkeeper.mledger.impl.NullLedgerOffloader; +import org.apache.bookkeeper.mledger.impl.NonAppendableLedgerOffloader; import org.apache.bookkeeper.mledger.util.Futures; import org.apache.commons.collections4.MapUtils; import org.apache.commons.lang3.StringUtils; @@ -2018,29 +2018,26 @@ private CompletableFuture getManagedLedgerConfig(@Nonnull T topicLevelOffloadPolicies, OffloadPoliciesImpl.oldPoliciesCompatible(nsLevelOffloadPolicies, policies.orElse(null)), getPulsar().getConfig().getProperties()); - if (NamespaceService.isSystemServiceNamespace(namespace.toString()) - || SystemTopicNames.isSystemTopic(topicName)) { - /* - Avoid setting broker internal system topics using off-loader because some of them are the - preconditions of other topics. The slow replying log speed will cause a delay in all the topic - loading.(timeout) - */ - managedLedgerConfig.setLedgerOffloader(NullLedgerOffloader.INSTANCE); - } else { - if (topicLevelOffloadPolicies != null) { - try { - LedgerOffloader topicLevelLedgerOffLoader = - pulsar().createManagedLedgerOffloader(offloadPolicies); - managedLedgerConfig.setLedgerOffloader(topicLevelLedgerOffLoader); - } catch (PulsarServerException e) { - throw new RuntimeException(e); - } - } else { - //If the topic level policy is null, use the namespace level - managedLedgerConfig - .setLedgerOffloader(pulsar.getManagedLedgerOffloader(namespace, offloadPolicies)); + if (topicLevelOffloadPolicies != null) { + try { + LedgerOffloader topicLevelLedgerOffLoader = pulsar().createManagedLedgerOffloader(offloadPolicies); + managedLedgerConfig.setLedgerOffloader(topicLevelLedgerOffLoader); + } catch (PulsarServerException e) { + throw new RuntimeException(e); } + } else { + //If the topic level policy is null, use the namespace level + managedLedgerConfig + .setLedgerOffloader(pulsar.getManagedLedgerOffloader(namespace, offloadPolicies)); } + if (managedLedgerConfig.getLedgerOffloader() != null + && managedLedgerConfig.getLedgerOffloader().isAppendable() + && (NamespaceService.isSystemServiceNamespace(namespace.toString()) + || SystemTopicNames.isSystemTopic(topicName))) { + managedLedgerConfig.setLedgerOffloader( + new NonAppendableLedgerOffloader(managedLedgerConfig.getLedgerOffloader())); + } + managedLedgerConfig.setTriggerOffloadOnTopicLoad(serviceConfig.isTriggerOffloadOnTopicLoad()); managedLedgerConfig.setDeletionAtBatchIndexLevelEnabled( diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiOffloadTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiOffloadTest.java index eac816bd81089a..1ea29c9d431bd0 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiOffloadTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiOffloadTest.java @@ -126,6 +126,7 @@ private void testOffload(String topicName, String mlName) throws Exception { CompletableFuture promise = new CompletableFuture<>(); doReturn(promise).when(offloader).offload(any(), any(), any()); + doReturn(true).when(offloader).isAppendable(); MessageId currentId = MessageId.latest; try (Producer p = pulsarClient.newProducer().topic(topicName).enableBatching(false).create()) { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java index aa236e09da99d3..2f27d5917f025f 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java @@ -76,6 +76,7 @@ import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl; import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.bookkeeper.mledger.impl.NullLedgerOffloader; +import org.apache.bookkeeper.mledger.impl.NonAppendableLedgerOffloader; import org.apache.http.HttpResponse; import org.apache.http.client.methods.HttpGet; import org.apache.http.impl.client.CloseableHttpClient; @@ -1883,6 +1884,10 @@ public void close() { final String namespace = "prop/" + UUID.randomUUID(); admin.namespaces().createNamespace(namespace); admin.namespaces().setOffloadPolicies(namespace, offloadPolicies); + Awaitility.await().untilAsserted(() -> { + OffloadPolicies policiesGot = admin.namespaces().getOffloadPolicies(namespace); + assertNotNull(policiesGot); + }); // Inject the cache to avoid real load off-loader jar final Map ledgerOffloaderMap = pulsar.getLedgerOffloaderMap(); @@ -1896,8 +1901,20 @@ public void close() { // (2) test system topic for (String eventTopicName : SystemTopicNames.EVENTS_TOPIC_NAMES) { - managedLedgerConfig = brokerService.getManagedLedgerConfig(TopicName.get(eventTopicName)).join(); - Assert.assertEquals(managedLedgerConfig.getLedgerOffloader(), NullLedgerOffloader.INSTANCE); + boolean offloadPoliciesExists = false; + try { + OffloadPolicies policiesGot = + admin.namespaces().getOffloadPolicies(TopicName.get(eventTopicName).getNamespace()); + offloadPoliciesExists = policiesGot != null; + } catch (PulsarAdminException.NotFoundException notFoundException) { + offloadPoliciesExists = false; + } + var managedLedgerConfig2 = brokerService.getManagedLedgerConfig(TopicName.get(eventTopicName)).join(); + if (offloadPoliciesExists) { + Assert.assertTrue(managedLedgerConfig2.getLedgerOffloader() instanceof NonAppendableLedgerOffloader); + } else { + Assert.assertEquals(managedLedgerConfig2.getLedgerOffloader(), NullLedgerOffloader.INSTANCE); + } } } }