diff --git a/conf/broker.conf b/conf/broker.conf index f28564a154326..5c1485aa9e9e2 100644 --- a/conf/broker.conf +++ b/conf/broker.conf @@ -1248,6 +1248,10 @@ managedLedgerMaxLedgerRolloverTimeMinutes=240 # Disable rollover with value 0 (Default value 0) managedLedgerInactiveLedgerRolloverTimeSeconds=0 +# Time to evict inactive offloaded ledger for inactive topic +# Disable eviction with value 0 +managedLedgerInactiveOffloadedLedgerEvictionTimeSeconds=600 + # Maximum ledger size before triggering a rollover for a topic (MB) managedLedgerMaxSizePerLedgerMbytes=2048 diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java index 6fc39170e851a..c1cbf0ad704fe 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java @@ -80,6 +80,7 @@ public class ManagedLedgerConfig { private ManagedLedgerInterceptor managedLedgerInterceptor; private Map properties; private int inactiveLedgerRollOverTimeMs = 0; + private long inactiveOffloadedLedgerEvictionTimeMs = 0; @Getter @Setter private boolean cacheEvictionByMarkDeletedPosition = false; @@ -715,6 +716,14 @@ public void setInactiveLedgerRollOverTime(int inactiveLedgerRollOverTimeMs, Time this.inactiveLedgerRollOverTimeMs = (int) unit.toMillis(inactiveLedgerRollOverTimeMs); } + public long getInactiveOffloadedLedgerEvictionTimeMs() { + return inactiveOffloadedLedgerEvictionTimeMs; + } + + public void setInactiveOffloadedLedgerEvictionTime(long inactiveOffloadedLedgerEvictionTime, TimeUnit unit) { + this.inactiveOffloadedLedgerEvictionTimeMs = unit.toMillis(inactiveOffloadedLedgerEvictionTime); + } + /** * Minimum cursors with backlog after which broker is allowed to cache read entries to reuse them for other cursors' * backlog reads. (Default = 0, broker will not cache backlog reads) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/OffloadedLedgerHandle.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/OffloadedLedgerHandle.java new file mode 100644 index 0000000000000..f45d115090f73 --- /dev/null +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/OffloadedLedgerHandle.java @@ -0,0 +1,29 @@ +/* + * 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; + +/** + * This is a marked interface for ledger handle that represent offloaded data. + */ +public interface OffloadedLedgerHandle { + + default long lastAccessTimestamp() { + return -1; + } +} 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 cb7fb2658c401..6f7fbe9edb1b6 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 @@ -117,6 +117,7 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException.NonRecoverableLedgerException; import org.apache.bookkeeper.mledger.ManagedLedgerException.TooManyRequestsException; import org.apache.bookkeeper.mledger.ManagedLedgerMXBean; +import org.apache.bookkeeper.mledger.OffloadedLedgerHandle; import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.PositionBound; import org.apache.bookkeeper.mledger.PositionFactory; @@ -343,6 +344,9 @@ public boolean isFenced() { @VisibleForTesting Map createdLedgerCustomMetadata; + private long lastEvictOffloadedLedgers; + private static final int MINIMUM_EVICTION_INTERVAL_DIVIDER = 10; + public ManagedLedgerImpl(ManagedLedgerFactoryImpl factory, BookKeeper bookKeeper, MetaStore store, ManagedLedgerConfig config, OrderedScheduler scheduledExecutor, final String name) { @@ -2078,6 +2082,8 @@ CompletableFuture getLedgerHandle(long ledgerId) { // TODO: improve this to load ledger offloader by driver name recorded in metadata Map offloadDriverMetadata = OffloadUtils.getOffloadDriverMetadata(info); offloadDriverMetadata.put("ManagedLedgerName", name); + log.info("[{}] Opening ledger {} from offload driver {} with uid {}", name, ledgerId, + config.getLedgerOffloader().getOffloadDriverName(), uid); openFuture = config.getLedgerOffloader().readOffloaded(ledgerId, uid, offloadDriverMetadata); } else { @@ -2103,11 +2109,20 @@ CompletableFuture getLedgerHandle(long ledgerId) { void invalidateReadHandle(long ledgerId) { CompletableFuture rhf = ledgerCache.remove(ledgerId); if (rhf != null) { - rhf.thenAccept(ReadHandle::closeAsync) - .exceptionally(ex -> { - log.warn("[{}] Failed to close a Ledger ReadHandle:", name, ex); - return null; - }); + rhf.thenCompose(r -> { + if (r instanceof OffloadedLedgerHandle) { + log.info("[{}] Closing ledger {} from offload driver {}", name, ledgerId, + config.getLedgerOffloader().getOffloadDriverName()); + } + return r.closeAsync().exceptionally(ex -> { + log.warn("[{}] Failed to close ledger {} ReadHandle with type {}", name, ledgerId, + r.getClass().getName(), ex); + return null; + }); + }).exceptionally(ex -> { + log.warn("[{}] Failed to close Ledger ReadHandle {}:", name, ledgerId, ex); + return null; + }); } } @@ -2716,7 +2731,56 @@ private Optional getOffloadPoliciesIfAppendable() { return Optional.ofNullable(ledgerOffloader.getOffloadPolicies()); } + @VisibleForTesting + synchronized List internalEvictOffloadedLedgers() { + long inactiveOffloadedLedgerEvictionTimeMs = config.getInactiveOffloadedLedgerEvictionTimeMs(); + if (inactiveOffloadedLedgerEvictionTimeMs <= 0) { + return Collections.emptyList(); + } + + long now = clock.millis(); + long minimumEvictionIntervalMs = inactiveOffloadedLedgerEvictionTimeMs / MINIMUM_EVICTION_INTERVAL_DIVIDER; + if (now - lastEvictOffloadedLedgers < minimumEvictionIntervalMs) { + // skip eviction if we have done it recently + return Collections.emptyList(); + } + + try { + List ledgersToRelease = new ArrayList<>(); + + ledgerCache.forEach((ledgerId, ledger) -> { + if (ledger.isDone() && !ledger.isCompletedExceptionally()) { + ReadHandle readHandle = ledger.join(); + if (readHandle instanceof OffloadedLedgerHandle) { + long lastAccessTimestamp = ((OffloadedLedgerHandle) readHandle).lastAccessTimestamp(); + if (lastAccessTimestamp >= 0) { + long delta = now - lastAccessTimestamp; + if (delta >= inactiveOffloadedLedgerEvictionTimeMs) { + log.info("[{}] Offloaded ledger {} can be released ({} ms elapsed since last access)", + name, ledgerId, delta); + ledgersToRelease.add(ledgerId); + } else if (log.isDebugEnabled()) { + log.debug( + "[{}] Offloaded ledger {} cannot be released ({} ms elapsed since last access)", + name, ledgerId, delta); + } + } + } + } + }); + for (Long ledgerId : ledgersToRelease) { + invalidateReadHandle(ledgerId); + } + return ledgersToRelease; + } finally { + lastEvictOffloadedLedgers = now; + } + } + void internalTrimLedgers(boolean isTruncate, CompletableFuture promise) { + + internalEvictOffloadedLedgers(); + if (!factory.isMetadataServiceAvailable()) { // Defer trimming of ledger if we cannot connect to metadata service promise.completeExceptionally(new MetaStoreException("Metadata service is not available")); diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadEvictUnusedLedgersTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadEvictUnusedLedgersTest.java new file mode 100644 index 0000000000000..c8f4602b53887 --- /dev/null +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadEvictUnusedLedgersTest.java @@ -0,0 +1,105 @@ +/* + * 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 static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotEquals; +import static org.testng.Assert.assertTrue; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Collectors; +import org.apache.bookkeeper.mledger.Entry; +import org.apache.bookkeeper.mledger.ManagedCursor; +import org.apache.bookkeeper.mledger.ManagedLedgerConfig; +import org.apache.bookkeeper.mledger.PositionFactory; +import org.apache.bookkeeper.test.MockedBookKeeperTestCase; +import org.awaitility.Awaitility; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testng.annotations.Test; + +public class OffloadEvictUnusedLedgersTest extends MockedBookKeeperTestCase { + private static final Logger log = LoggerFactory.getLogger(OffloadEvictUnusedLedgersTest.class); + + @Test + public void testEvictUnusedLedgers() throws Exception { + OffloadPrefixReadTest.MockLedgerOffloader offloader = + new OffloadPrefixReadTest.MockLedgerOffloader(); + ManagedLedgerConfig config = new ManagedLedgerConfig(); + config.setMaxEntriesPerLedger(10); + config.setMinimumRolloverTime(0, TimeUnit.SECONDS); + config.setRetentionTime(10, TimeUnit.MINUTES); + config.setRetentionSizeInMB(10); + long inactiveOffloadedLedgerEvictionTimeMs = 1000; + config.setInactiveOffloadedLedgerEvictionTime(inactiveOffloadedLedgerEvictionTimeMs, TimeUnit.MILLISECONDS); + config.setLedgerOffloader(offloader); + ManagedLedgerImpl ledger = (ManagedLedgerImpl)factory.open("my_test_ledger_evict", config); + + // no evict when no offloaded ledgers + assertTrue(ledger.internalEvictOffloadedLedgers().isEmpty()); + + int i = 0; + for (; i < 25; i++) { + String content = "entry-" + i; + ledger.addEntry(content.getBytes()); + } + assertEquals(ledger.getLedgersInfoAsList().size(), 3); + + ledger.offloadPrefix(ledger.getLastConfirmedEntry()); + + assertEquals(ledger.getLedgersInfoAsList().size(), 3); + assertEquals(ledger.getLedgersInfoAsList().stream() + .filter(e -> e.getOffloadContext().getComplete()) + .map(e -> e.getLedgerId()).collect(Collectors.toSet()), + offloader.offloadedLedgers()); + + // ledgers should be marked as offloaded + ledger.getLedgersInfoAsList().stream().allMatch(l -> l.hasOffloadContext()); + + // no evict when no offloaded ledgers are marked as inactive + assertTrue(ledger.internalEvictOffloadedLedgers().isEmpty()); + + ManagedCursor cursor = ledger.newNonDurableCursor(PositionFactory.EARLIEST); + int j = 0; + for (Entry e : cursor.readEntries(25)) { + assertEquals(new String(e.getData()), "entry-" + j++); + } + cursor.close(); + + // set last access time to be 2x inactiveOffloadedLedgerEvictionTimeMs + AtomicLong first = new AtomicLong(-1); + assertTrue(!ledger.ledgerCache.isEmpty()); + ledger.ledgerCache.forEach((id, l) -> { + if (first.compareAndSet(-1, id)) { + OffloadPrefixReadTest.MockOffloadReadHandle handle = + (OffloadPrefixReadTest.MockOffloadReadHandle) l.join(); + handle.setLastAccessTimestamp(System.currentTimeMillis() - inactiveOffloadedLedgerEvictionTimeMs * 2); + } + }); + assertNotEquals(first.get(), -1L); + + Awaitility.await().untilAsserted(() -> { + List evicted = ledger.internalEvictOffloadedLedgers(); + assertEquals(evicted.size(), 1); + assertEquals(first.get(), evicted.get(0).longValue()); + }); + } + +} 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 48751417e1714..9255cb188de2b 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 @@ -31,17 +31,18 @@ import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertTrue; import io.netty.buffer.ByteBuf; - import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Map; import java.util.NavigableMap; +import java.util.Set; import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; import lombok.SneakyThrows; import org.apache.bookkeeper.client.api.DigestType; import org.apache.bookkeeper.client.api.LastConfirmedAndEntry; @@ -57,6 +58,7 @@ import org.apache.bookkeeper.mledger.ManagedLedgerConfig; import org.apache.bookkeeper.mledger.ManagedLedgerException; import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig; +import org.apache.bookkeeper.mledger.OffloadedLedgerHandle; import org.apache.bookkeeper.mledger.PositionFactory; import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo; import org.apache.bookkeeper.mledger.util.MockClock; @@ -313,6 +315,10 @@ static class MockLedgerOffloader implements LedgerOffloader { OffloadPoliciesImpl.DEFAULT_OFFLOAD_DELETION_LAG_IN_MILLIS, OffloadPoliciesImpl.DEFAULT_OFFLOADED_READ_PRIORITY); + Set offloadedLedgers() { + return offloads.values().stream().map(ReadHandle::getId).collect(Collectors.toSet()); + } + @Override public String getOffloadDriverName() { @@ -373,10 +379,11 @@ public CompletableFuture closeAsync() { } } - static class MockOffloadReadHandle implements ReadHandle { + static class MockOffloadReadHandle implements ReadHandle, OffloadedLedgerHandle { final long id; final List entries = new ArrayList(); final LedgerMetadata metadata; + long lastAccessTimestamp = System.currentTimeMillis(); MockOffloadReadHandle(ReadHandle toCopy) throws Exception { id = toCopy.getId(); @@ -454,6 +461,15 @@ private CompletableFuture unsupported() { future.completeExceptionally(new UnsupportedOperationException()); return future; } + + @Override + public long lastAccessTimestamp() { + return lastAccessTimestamp; + } + + public void setLastAccessTimestamp(long lastAccessTimestamp) { + this.lastAccessTimestamp = lastAccessTimestamp; + } } static class MockMetadata implements LedgerMetadata { diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java index 9ab3f533909e0..af6e2d1418059 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java @@ -3415,6 +3415,14 @@ public double getLoadBalancerBandwidthOutResourceWeight() { ) private int managedLedgerInactiveLedgerRolloverTimeSeconds = 0; + @FieldContext( + dynamic = true, + category = CATEGORY_STORAGE_ML, + doc = "Time to evict inactive offloaded ledger for inactive topic. " + + "Disable eviction with value 0 (Default value 600)" + ) + private int managedLedgerInactiveOffloadedLedgerEvictionTimeSeconds = 600; + @FieldContext( category = CATEGORY_STORAGE_ML, doc = "Evicting cache data by the slowest markDeletedPosition or readPosition. " 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 ab0bef938473b..40ed9b353340e 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 @@ -2089,6 +2089,10 @@ public CompletableFuture getManagedLedgerConfig(@NonNull To managedLedgerConfig.setLazyCursorRecovery(serviceConfig.isLazyCursorRecovery()); managedLedgerConfig.setInactiveLedgerRollOverTime( serviceConfig.getManagedLedgerInactiveLedgerRolloverTimeSeconds(), TimeUnit.SECONDS); + managedLedgerConfig.setInactiveOffloadedLedgerEvictionTime( + serviceConfig.getManagedLedgerInactiveOffloadedLedgerEvictionTimeSeconds(), + TimeUnit.SECONDS); + managedLedgerConfig.setCacheEvictionByMarkDeletedPosition( serviceConfig.isCacheEvictionByMarkDeletedPosition()); managedLedgerConfig.setMinimumBacklogCursorsForCaching( diff --git a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java index e050d74a332bc..1f2f901f51426 100644 --- a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java +++ b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java @@ -40,6 +40,7 @@ import org.apache.bookkeeper.client.impl.LedgerEntryImpl; import org.apache.bookkeeper.mledger.LedgerOffloaderStats; import org.apache.bookkeeper.mledger.ManagedLedgerException; +import org.apache.bookkeeper.mledger.OffloadedLedgerHandle; import org.apache.bookkeeper.mledger.offload.jcloud.BackedInputStream; import org.apache.bookkeeper.mledger.offload.jcloud.OffloadIndexBlock; import org.apache.bookkeeper.mledger.offload.jcloud.OffloadIndexBlockBuilder; @@ -52,7 +53,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public class BlobStoreBackedReadHandleImpl implements ReadHandle { +public class BlobStoreBackedReadHandleImpl implements ReadHandle, OffloadedLedgerHandle { private static final Logger log = LoggerFactory.getLogger(BlobStoreBackedReadHandleImpl.class); private final long ledgerId; @@ -70,6 +71,8 @@ enum State { private volatile State state = null; + private volatile long lastAccessTimestamp = System.currentTimeMillis(); + private BlobStoreBackedReadHandleImpl(long ledgerId, OffloadIndexBlock index, BackedInputStream inputStream, ExecutorService executor, OffsetsCache entryOffsetsCache) { @@ -119,7 +122,9 @@ public CompletableFuture readAsync(long firstEntry, long lastEntr getId(), firstEntry, lastEntry, (1 + lastEntry - firstEntry)); } CompletableFuture promise = new CompletableFuture<>(); + touch(); executor.execute(() -> { + touch(); if (state == State.Closed) { log.warn("Reading a closed read handler. Ledger ID: {}, Read range: {}-{}", ledgerId, firstEntry, lastEntry); @@ -208,6 +213,7 @@ public CompletableFuture readAsync(long firstEntry, long lastEntr } private void seekToEntry(long nextExpectedId) throws IOException { + touch(); Long knownOffset = entryOffsetsCache.getIfPresent(ledgerId, nextExpectedId); if (knownOffset != null) { inputStream.seek(knownOffset); @@ -312,4 +318,13 @@ public static ReadHandle open(ScheduledExecutorService executor, State getState() { return this.state; } + + @Override + public long lastAccessTimestamp() { + return lastAccessTimestamp; + } + + private void touch() { + lastAccessTimestamp = System.currentTimeMillis(); + } }