diff --git a/conf/broker.conf b/conf/broker.conf index 856d173d6bb33..fb2cb1e36e04b 100644 --- a/conf/broker.conf +++ b/conf/broker.conf @@ -440,6 +440,16 @@ managedLedgerCacheSizeMB= # Threshold to which bring down the cache level when eviction is triggered managedLedgerCacheEvictionWatermark=0.9 +# Configure the cache eviction frequency for the managed ledger cache (evictions/sec) +managedLedgerCacheEvictionFrequency=100.0 + +# All entries that have stayed in cache for more than the configured time, will be evicted +managedLedgerCacheEvictionTimeThresholdMillis=1000 + +# Configure the threshold (in number of entries) from where a cursor should be considered 'backlogged' +# and thus should be set as inactive. +managedLedgerCursorBackloggedThreshold=1000 + # Rate limit the amount of writes per second generated by consumer acking the messages managedLedgerDefaultMarkDeleteRateLimit=1.0 diff --git a/conf/standalone.conf b/conf/standalone.conf index 4cbf931c75748..b8f13c51a0489 100644 --- a/conf/standalone.conf +++ b/conf/standalone.conf @@ -301,6 +301,16 @@ managedLedgerCacheSizeMB= # Threshold to which bring down the cache level when eviction is triggered managedLedgerCacheEvictionWatermark=0.9 +# Configure the cache eviction frequency for the managed ledger cache (evictions/sec) +managedLedgerCacheEvictionFrequency=100.0 + +# All entries that have stayed in cache for more than the configured time, will be evicted +managedLedgerCacheEvictionTimeThresholdMillis=1000 + +# Configure the threshold (in number of entries) from where a cursor should be considered 'backlogged' +# and thus should be set as inactive. +managedLedgerCursorBackloggedThreshold=1000 + # Rate limit the amount of writes generated by consumer acking the messages managedLedgerDefaultMarkDeleteRateLimit=0.1 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 8f890505167fe..4a57055393c67 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 @@ -515,9 +515,9 @@ public ManagedLedgerConfig setClock(Clock clock) { } /** - * + * * Ledger-Op (Create/Delete) timeout - * + * * @return */ public long getMetadataOperationsTimeoutSeconds() { @@ -526,17 +526,17 @@ public long getMetadataOperationsTimeoutSeconds() { /** * Ledger-Op (Create/Delete) timeout after which callback will be completed with failure - * + * * @param metadataOperationsTimeoutSeconds */ public ManagedLedgerConfig setMetadataOperationsTimeoutSeconds(long metadataOperationsTimeoutSeconds) { this.metadataOperationsTimeoutSeconds = metadataOperationsTimeoutSeconds; return this; } - + /** * Ledger read-entry timeout - * + * * @return */ public long getReadEntryTimeoutSeconds() { @@ -546,7 +546,7 @@ public long getReadEntryTimeoutSeconds() { /** * Ledger read entry timeout after which callback will be completed with failure. (disable timeout by setting * readTimeoutSeconds <= 0) - * + * * @param readTimeoutSeconds * @return */ @@ -554,14 +554,14 @@ public ManagedLedgerConfig setReadEntryTimeoutSeconds(long readEntryTimeoutSecon this.readEntryTimeoutSeconds = readEntryTimeoutSeconds; return this; } - + public long getAddEntryTimeoutSeconds() { return addEntryTimeoutSeconds; } /** * Add-entry timeout after which add-entry callback will be failed if add-entry is not succeeded. - * + * * @param addEntryTimeoutSeconds */ public ManagedLedgerConfig setAddEntryTimeoutSeconds(long addEntryTimeoutSeconds) { diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerFactoryConfig.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerFactoryConfig.java index 81673f4351ba9..40b815df76613 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerFactoryConfig.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerFactoryConfig.java @@ -37,7 +37,18 @@ public class ManagedLedgerFactoryConfig { private int numManagedLedgerWorkerThreads = Runtime.getRuntime().availableProcessors(); private int numManagedLedgerSchedulerThreads = Runtime.getRuntime().availableProcessors(); - public long getMaxCacheSize() { - return maxCacheSize; - } + /** + * Frequency of cache eviction triggering. Default is 100 times per second. + */ + private double cacheEvictionFrequency = 100; + + /** + * All entries that have stayed in cache for more than the configured time, will be evicted + */ + private long cacheEvictionTimeThresholdMillis = 1000; + + /** + * Threshould to consider a cursor as "backlogged" + */ + private long thresholdBackloggedCursor = 1000; } diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCache.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCache.java index 0c99650cecf18..0e20a0facffb2 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCache.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCache.java @@ -50,10 +50,12 @@ public interface EntryCache extends Comparable { * Remove from cache all the entries related to a ledger up to lastPosition included. * * @param lastPosition - * the position of the last entry to be invalidated (inclusive) + * the position of the last entry to be invalidated (non-inclusive) */ void invalidateEntries(PositionImpl lastPosition); + void invalidateEntriesBeforeTimestamp(long timestamp); + /** * Remove from the cache all the entries belonging to a specific ledger. * diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheImpl.java index a435c1442b90b..053e9002cc977 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheImpl.java @@ -21,15 +21,17 @@ import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkNotNull; import static org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.createManagedLedgerException; -import static org.apache.bookkeeper.mledger.util.SafeRun.safeRun; import com.google.common.collect.Lists; import com.google.common.primitives.Longs; + import io.netty.buffer.ByteBuf; import io.netty.buffer.PooledByteBufAllocator; + import java.util.Collection; import java.util.Iterator; import java.util.List; + import org.apache.bookkeeper.client.api.BKException; import org.apache.bookkeeper.client.api.LedgerEntry; import org.apache.bookkeeper.client.api.ReadHandle; @@ -37,7 +39,6 @@ import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback; import org.apache.bookkeeper.mledger.ManagedLedgerException; import org.apache.bookkeeper.mledger.util.RangeCache; -import org.apache.bookkeeper.mledger.util.RangeCache.Weighter; import org.apache.commons.lang3.tuple.Pair; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -53,12 +54,10 @@ public class EntryCacheImpl implements EntryCache { private static final double MB = 1024 * 1024; - private static final Weighter entryWeighter = EntryImpl::getLength; - public EntryCacheImpl(EntryCacheManager manager, ManagedLedgerImpl ml) { this.manager = manager; this.ml = ml; - this.entries = new RangeCache<>(entryWeighter); + this.entries = new RangeCache<>(EntryImpl::getLength, EntryImpl::getTimestamp); if (log.isDebugEnabled()) { log.debug("[{}] Initialized managed-ledger entry cache", ml.getName()); @@ -132,7 +131,7 @@ public boolean insert(EntryImpl entry) { public void invalidateEntries(final PositionImpl lastPosition) { final PositionImpl firstPosition = PositionImpl.get(-1, 0); - Pair removed = entries.removeRange(firstPosition, lastPosition, true); + Pair removed = entries.removeRange(firstPosition, lastPosition, false); int entriesRemoved = removed.getLeft(); long sizeRemoved = removed.getRight(); if (log.isDebugEnabled()) { @@ -173,7 +172,7 @@ public void asyncReadEntry(ReadHandle lh, PositionImpl position, final ReadEntry callback.readEntryFailed(createManagedLedgerException(t), ctx); } } - + private void asyncReadEntry0(ReadHandle lh, PositionImpl position, final ReadEntryCallback callback, final Object ctx) { if (log.isDebugEnabled()) { @@ -229,7 +228,7 @@ public void asyncReadEntry(ReadHandle lh, long firstEntry, long lastEntry, boole callback.readEntriesFailed(createManagedLedgerException(t), ctx); } } - + @SuppressWarnings({ "unchecked", "rawtypes" }) private void asyncReadEntry0(ReadHandle lh, long firstEntry, long lastEntry, boolean isSlowestReader, final ReadEntriesCallback callback, Object ctx) { @@ -341,5 +340,11 @@ public Pair evictEntries(long sizeToFree) { return evicted; } + @Override + public void invalidateEntriesBeforeTimestamp(long timestamp) { + long evictedSize = entries.evictLEntriesBeforeTimestamp(timestamp); + manager.entriesRemoved(evictedSize); + } + private static final Logger log = LoggerFactory.getLogger(EntryCacheImpl.class); } diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheManager.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheManager.java index c551002b474f9..526796e3c26bd 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheManager.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheManager.java @@ -189,6 +189,10 @@ public Pair evictEntries(long sizeToFree) { return Pair.of(0, (long) 0); } + @Override + public void invalidateEntriesBeforeTimestamp(long timestamp) { + } + @Override public void asyncReadEntry(ReadHandle lh, long firstEntry, long lastEntry, boolean isSlowestReader, final ReadEntriesCallback callback, Object ctx) { diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java index eeccfe7857d2c..58ea09923e64e 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java @@ -40,12 +40,14 @@ protected EntryImpl newObject(Handle handle) { }; private final Handle recyclerHandle; + private long timestamp; private long ledgerId; private long entryId; ByteBuf data; public static EntryImpl create(LedgerEntry ledgerEntry) { EntryImpl entry = RECYCLER.get(); + entry.timestamp = System.nanoTime(); entry.ledgerId = ledgerEntry.getLedgerId(); entry.entryId = ledgerEntry.getEntryId(); entry.data = ledgerEntry.getEntryBuffer(); @@ -57,6 +59,7 @@ public static EntryImpl create(LedgerEntry ledgerEntry) { // Used just for tests public static EntryImpl create(long ledgerId, long entryId, byte[] data) { EntryImpl entry = RECYCLER.get(); + entry.timestamp = System.nanoTime(); entry.ledgerId = ledgerId; entry.entryId = entryId; entry.data = Unpooled.wrappedBuffer(data); @@ -66,6 +69,7 @@ public static EntryImpl create(long ledgerId, long entryId, byte[] data) { public static EntryImpl create(long ledgerId, long entryId, ByteBuf data) { EntryImpl entry = RECYCLER.get(); + entry.timestamp = System.nanoTime(); entry.ledgerId = ledgerId; entry.entryId = entryId; entry.data = data; @@ -76,6 +80,7 @@ public static EntryImpl create(long ledgerId, long entryId, ByteBuf data) { public static EntryImpl create(PositionImpl position, ByteBuf data) { EntryImpl entry = RECYCLER.get(); + entry.timestamp = System.nanoTime(); entry.ledgerId = position.getLedgerId(); entry.entryId = position.getEntryId(); entry.data = data; @@ -86,6 +91,7 @@ public static EntryImpl create(PositionImpl position, ByteBuf data) { public static EntryImpl create(EntryImpl other) { EntryImpl entry = RECYCLER.get(); + entry.timestamp = System.nanoTime(); entry.ledgerId = other.ledgerId; entry.entryId = other.entryId; entry.data = other.data.retainedDuplicate(); @@ -97,6 +103,10 @@ private EntryImpl(Recycler.Handle recyclerHandle) { this.recyclerHandle = recyclerHandle; } + public long getTimestamp() { + return timestamp; + } + @Override public ByteBuf getDataBuffer() { return data; @@ -152,6 +162,7 @@ protected void deallocate() { // This method is called whenever the ref-count of the EntryImpl reaches 0, so that now we can recycle it data.release(); data = null; + timestamp = -1; ledgerId = -1; entryId = -1; recyclerHandle.recycle(this); diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java index b9f63a850b9e9..7399233f1694f 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java @@ -24,6 +24,8 @@ import com.google.common.base.Predicates; import com.google.common.collect.Maps; +import io.netty.util.concurrent.DefaultThreadFactory; + import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -32,6 +34,8 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentSkipListMap; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -82,6 +86,8 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory { protected final OrderedScheduler scheduledExecutor; private final OrderedExecutor orderedExecutor; + private final ExecutorService cacheEvictionExecutor; + protected final ManagedLedgerFactoryMBeanImpl mbean; protected final ConcurrentHashMap> ledgers = new ConcurrentHashMap<>(); @@ -89,6 +95,9 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory { private long lastStatTimestamp = System.nanoTime(); private final ScheduledFuture statsTask; + + private final long cacheEvictionTimeThresholdNanos; + private static final int StatsPeriodSeconds = 60; public ManagedLedgerFactoryImpl(ClientConfiguration bkClientConfiguration) throws Exception { @@ -130,6 +139,8 @@ private ManagedLedgerFactoryImpl(BookKeeper bookKeeper, boolean isBookkeeperMana .numThreads(config.getNumManagedLedgerWorkerThreads()) .name("bookkeeper-ml-workers") .build(); + cacheEvictionExecutor = Executors + .newSingleThreadExecutor(new DefaultThreadFactory("bookkeeper-ml-cache-eviction")); this.bookKeeper = bookKeeper; this.isBookkeeperManaged = isBookkeeperManaged; @@ -139,6 +150,13 @@ private ManagedLedgerFactoryImpl(BookKeeper bookKeeper, boolean isBookkeeperMana this.mbean = new ManagedLedgerFactoryMBeanImpl(this); this.entryCacheManager = new EntryCacheManager(this); this.statsTask = scheduledExecutor.scheduleAtFixedRate(() -> refreshStats(), 0, StatsPeriodSeconds, TimeUnit.SECONDS); + + + this.cacheEvictionTimeThresholdNanos = TimeUnit.MILLISECONDS + .toNanos(config.getCacheEvictionTimeThresholdMillis()); + + + cacheEvictionExecutor.execute(this::cacheEvictionTask); } private synchronized void refreshStats() { @@ -147,15 +165,48 @@ private synchronized void refreshStats() { mbean.refreshStats(period, TimeUnit.NANOSECONDS); ledgers.values().forEach(mlfuture -> { - ManagedLedgerImpl ml = mlfuture.getNow(null); - if (ml != null) { - ml.mbean.refreshStats(period, TimeUnit.NANOSECONDS); + if (mlfuture.isDone() && !mlfuture.isCompletedExceptionally()) { + ManagedLedgerImpl ml = mlfuture.getNow(null); + if (ml != null) { + ml.mbean.refreshStats(period, TimeUnit.NANOSECONDS); + } } }); lastStatTimestamp = now; } + private void cacheEvictionTask() { + double evictionFrequency = Math.max(Math.min(config.getCacheEvictionFrequency(), 1000.0), 0.001); + long waitTimeMillis = (long) (1000 / evictionFrequency); + + while (true) { + try { + doCacheEviction(); + + Thread.sleep(waitTimeMillis); + } catch (InterruptedException e) { + // Factory is shutting down + return; + } catch (Throwable t) { + log.warn("Exception while performing cache eviction: {}", t.getMessage(), t); + } + } + } + + private synchronized void doCacheEviction() { + long maxTimestamp = System.nanoTime() - cacheEvictionTimeThresholdNanos; + + ledgers.values().forEach(mlfuture -> { + if (mlfuture.isDone() && !mlfuture.isCompletedExceptionally()) { + ManagedLedgerImpl ml = mlfuture.getNow(null); + if (ml != null) { + ml.doCacheEviction(maxTimestamp); + } + } + }); + } + /** * Helper for getting stats. * @@ -366,6 +417,7 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) { scheduledExecutor.shutdown(); orderedExecutor.shutdown(); + cacheEvictionExecutor.shutdownNow(); entryCacheManager.clear(); } 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 940fb7ffde291..6b9f529c9e58c 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 @@ -20,10 +20,25 @@ import static com.google.common.base.Preconditions.checkArgument; import static java.lang.Math.min; +import static org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.FALSE; +import static org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.TRUE; import static org.apache.bookkeeper.mledger.util.SafeRun.safeRun; +import com.google.common.collect.BoundType; +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; +import com.google.common.collect.Queues; +import com.google.common.collect.Range; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.util.Recycler; +import io.netty.util.Recycler.Handle; + import java.time.Clock; import java.util.Collections; +import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; @@ -62,13 +77,13 @@ import org.apache.bookkeeper.common.util.OrderedExecutor; import org.apache.bookkeeper.common.util.OrderedScheduler; import org.apache.bookkeeper.common.util.Retries; -import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.CloseCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCursorCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteLedgerCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.OffloadCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.OpenCursorCallback; +import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.TerminateCallback; import org.apache.bookkeeper.mledger.Entry; @@ -99,35 +114,15 @@ import org.apache.bookkeeper.mledger.util.CallbackMutex; import org.apache.bookkeeper.mledger.util.Futures; import org.apache.commons.lang3.tuple.Pair; -import org.apache.pulsar.common.api.Commands; import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.InitialPosition; -import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata; import org.apache.pulsar.common.util.collections.ConcurrentLongHashMap; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.google.common.collect.BoundType; -import com.google.common.collect.ImmutableMap; -import com.google.common.collect.Lists; -import com.google.common.collect.Maps; -import com.google.common.collect.Queues; -import com.google.common.collect.Range; -import com.google.common.util.concurrent.RateLimiter; - -import io.netty.buffer.ByteBuf; -import io.netty.buffer.Unpooled; -import io.netty.util.Recycler; -import io.netty.util.Recycler.Handle; -import java.util.HashMap; -import static org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.TRUE; -import static org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.FALSE; - public class ManagedLedgerImpl implements ManagedLedger, CreateCallback { private final static long MegaByte = 1024 * 1024; protected final static int AsyncOperationTimeoutSeconds = 30; - private final static long maxActiveCursorBacklogEntries = 100; - private static long maxMessageCacheRetentionTimeMillis = 10 * 1000; protected final BookKeeper bookKeeper; protected final String name; @@ -160,8 +155,6 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback { @SuppressWarnings("unused") private volatile long totalSize = 0; - private RateLimiter updateCursorRateLimit; - // Cursors that are waiting to be notified when new entries are persisted final ConcurrentLinkedQueue waitingCursors; @@ -235,6 +228,9 @@ enum PositionBound { private static final AtomicLongFieldUpdater READ_OP_COUNT_UPDATER = AtomicLongFieldUpdater .newUpdater(ManagedLedgerImpl.class, "readOpCount"); private volatile long readOpCount = 0; + + private final long backloggedCursorThresholdEntries; + // last read-operation's callback to check read-timeout on it. private volatile ReadEntryCallbackWrapper lastReadCallback = null; @@ -267,8 +263,8 @@ public ManagedLedgerImpl(ManagedLedgerFactoryImpl factory, BookKeeper bookKeeper this.entryCache = factory.getEntryCacheManager().getEntryCache(this); this.waitingCursors = Queues.newConcurrentLinkedQueue(); this.uninitializedCursors = Maps.newHashMap(); - this.updateCursorRateLimit = RateLimiter.create(1); this.clock = config.getClock(); + this.backloggedCursorThresholdEntries = factory.getConfig().getThresholdBackloggedCursor(); // Get the next rollover time. Add a random value upto 5% to avoid rollover multiple ledgers at the same time this.maximumRolloverTimeMs = (long) (config.getMaximumRolloverTimeMs() * (1 + random.nextDouble() * 5 / 100.0)); @@ -904,63 +900,14 @@ public long getTotalSize() { @Override public void checkBackloggedCursors() { - // activate caught up cursors cursors.forEach(cursor -> { - if (cursor.getNumberOfEntries() < maxActiveCursorBacklogEntries) { + if (cursor.getNumberOfEntries() < backloggedCursorThresholdEntries) { cursor.setActive(); + } else { + cursor.setInactive(); } }); - - // deactivate backlog cursors - Iterator cursors = activeCursors.iterator(); - while (cursors.hasNext()) { - ManagedCursor cursor = cursors.next(); - long backlogEntries = cursor.getNumberOfEntries(); - if (backlogEntries > maxActiveCursorBacklogEntries) { - PositionImpl readPosition = (PositionImpl) cursor.getReadPosition(); - readPosition = isValidPosition(readPosition) ? readPosition : getNextValidPosition(readPosition); - if (readPosition == null) { - if (log.isDebugEnabled()) { - log.debug("[{}] Couldn't find valid read position [{}] {}", name, cursor.getName(), - cursor.getReadPosition()); - } - continue; - } - try { - asyncReadEntry(readPosition, new ReadEntryCallback() { - - @Override - public void readEntryFailed(ManagedLedgerException e, Object ctx) { - log.warn("[{}] Failed while reading entries on [{}] {}", name, cursor.getName(), - e.getMessage(), e); - - } - - @Override - public void readEntryComplete(Entry entry, Object ctx) { - MessageMetadata msgMetadata = null; - try { - msgMetadata = Commands.parseMessageMetadata(entry.getDataBuffer()); - long msgTimeSincePublish = (clock.millis() - msgMetadata.getPublishTime()); - if (msgTimeSincePublish > maxMessageCacheRetentionTimeMillis) { - cursor.setInactive(); - } - } finally { - if (msgMetadata != null) { - msgMetadata.recycle(); - } - entry.release(); - } - - } - }, null); - } catch (Exception e) { - log.warn("[{}] Failed while reading entries from cache on [{}] {}", name, cursor.getName(), - e.getMessage(), e); - } - } - } } @Override @@ -1578,13 +1525,6 @@ private void internalReadFromLedger(ReadHandle ledger, OpReadEntry opReadEntry) lastEntry); } asyncReadEntry(ledger, firstEntry, lastEntry, false, opReadEntry, opReadEntry.ctx); - - if (updateCursorRateLimit.tryAcquire()) { - if (isCursorActive(cursor)) { - final PositionImpl lastReadPosition = PositionImpl.get(ledger.getId(), lastEntry); - discardEntriesFromCache(cursor, lastReadPosition); - } - } } protected void asyncReadEntry(ReadHandle ledger, PositionImpl position, ReadEntryCallback callback, Object ctx) { @@ -1780,6 +1720,31 @@ void discardEntriesFromCache(ManagedCursorImpl cursor, PositionImpl newPosition) } } + void doCacheEviction(long maxTimestamp) { + // Always remove all entries already read by active cursors + PositionImpl slowestReaderPos = getEarlierReadPositionForActiveCursors(); + if (slowestReaderPos != null) { + entryCache.invalidateEntries(slowestReaderPos); + } + + // Remove entries older than the cutoff threshold + entryCache.invalidateEntriesBeforeTimestamp(maxTimestamp); + } + + private PositionImpl getEarlierReadPositionForActiveCursors() { + PositionImpl smallest = null; + for (ManagedCursor cursor : activeCursors) { + PositionImpl p = (PositionImpl) cursor.getReadPosition(); + if (smallest == null) { + smallest = p; + } else if (p.compareTo(smallest) < 0) { + smallest = p; + } + } + + return smallest; + } + void updateCursor(ManagedCursorImpl cursor, PositionImpl newPosition) { Pair pair = cursors.cursorUpdated(cursor, newPosition); if (pair == null) { diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/RangeCache.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/RangeCache.java index b9b3aceb0df9b..e64a40de8aabb 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/RangeCache.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/RangeCache.java @@ -43,12 +43,13 @@ public class RangeCache, Value extends ReferenceCoun private final ConcurrentNavigableMap entries; private AtomicLong size; // Total size of values stored in cache private final Weighter weighter; // Weighter object used to extract the size from values + private final TimestampExtractor timestampExtractor; // Extract the timestamp associated with a value /** * Construct a new RangeLruCache with default Weighter. */ public RangeCache() { - this(new DefaultWeighter()); + this(new DefaultWeighter(), (x) -> System.nanoTime()); } /** @@ -57,10 +58,11 @@ public RangeCache() { * @param weighter * a custom weighter to compute the size of each stored value */ - public RangeCache(Weighter weighter) { + public RangeCache(Weighter weighter, TimestampExtractor timestampExtractor) { this.size = new AtomicLong(0); this.entries = new ConcurrentSkipListMap<>(); this.weighter = weighter; + this.timestampExtractor = timestampExtractor; } /** @@ -175,6 +177,34 @@ public Pair evictLeastAccessedEntries(long minSize) { return Pair.of(removedEntries, removedSize); } + /** + * + * @param maxTimestamp the max timestamp of the entries to be evicted + * @return the tota + */ + public long evictLEntriesBeforeTimestamp(long maxTimestamp) { + long removedSize = 0; + + while (true) { + Map.Entry entry = entries.firstEntry(); + if (entry == null || timestampExtractor.getTimestamp(entry.getValue()) > maxTimestamp) { + break; + } + + entry = entries.pollFirstEntry(); + if (entry == null) { + break; + } + + Value value = entry.getValue(); + removedSize += weighter.getSize(value); + value.release(); + } + + size.addAndGet(-removedSize); + return removedSize; + } + /** * Just for testing. Getting the number of entries is very expensive on the conncurrent map */ @@ -217,6 +247,15 @@ public interface Weighter { long getSize(ValueT value); } + /** + * Interface of a object that is able to the extract the "timestamp" of the cached values. + * + * @param + */ + public interface TimestampExtractor { + long getTimestamp(ValueT value); + } + /** * Default cache weighter, every value is assumed the same cost. * diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/EntryCacheManagerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/EntryCacheManagerTest.java index 47c515d13757a..69c23e323439e 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/EntryCacheManagerTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/EntryCacheManagerTest.java @@ -28,6 +28,9 @@ import org.apache.bookkeeper.common.util.OrderedScheduler; import org.apache.bookkeeper.mledger.Entry; +import org.apache.bookkeeper.mledger.ManagedCursor; +import org.apache.bookkeeper.mledger.ManagedLedger; +import org.apache.bookkeeper.mledger.ManagedLedgerFactory; import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig; import org.apache.bookkeeper.test.MockedBookKeeperTestCase; import org.testng.annotations.BeforeClass; @@ -99,15 +102,15 @@ void simple() throws Exception { assertEquals(cacheManager.getSize(), 3); assertEquals(cache2.getSize(), 3); - // Should remove 2 entries + // Should remove 1 entry cache2.invalidateEntries(new PositionImpl(2, 1)); - assertEquals(cacheManager.getSize(), 1); - assertEquals(cache2.getSize(), 1); + assertEquals(cacheManager.getSize(), 2); + assertEquals(cache2.getSize(), 2); cacheManager.mlFactoryMBean.refreshStats(1, TimeUnit.SECONDS); assertEquals(cacheManager.mlFactoryMBean.getCacheMaxSize(), 10); - assertEquals(cacheManager.mlFactoryMBean.getCacheUsedSize(), 1); + assertEquals(cacheManager.mlFactoryMBean.getCacheUsedSize(), 2); assertEquals(cacheManager.mlFactoryMBean.getCacheHitsRate(), 0.0); assertEquals(cacheManager.mlFactoryMBean.getCacheMissesRate(), 0.0); assertEquals(cacheManager.mlFactoryMBean.getCacheHitsThroughput(), 0.0); @@ -207,6 +210,7 @@ void verifyHitsMisses() throws Exception { ManagedLedgerFactoryConfig config = new ManagedLedgerFactoryConfig(); config.setMaxCacheSize(7 * 10); config.setCacheEvictionWatermark(0.8); + config.setCacheEvictionFrequency(1); factory = new ManagedLedgerFactoryImpl(bkc, bkc.getZkHandle(), config); @@ -263,10 +267,47 @@ void verifyHitsMisses() throws Exception { entries.forEach(e -> e.release()); cacheManager.mlFactoryMBean.refreshStats(1, TimeUnit.SECONDS); - assertEquals(cacheManager.mlFactoryMBean.getCacheUsedSize(), 0); + assertEquals(cacheManager.mlFactoryMBean.getCacheUsedSize(), 7); assertEquals(cacheManager.mlFactoryMBean.getCacheHitsRate(), 0.0); assertEquals(cacheManager.mlFactoryMBean.getCacheMissesRate(), 0.0); assertEquals(cacheManager.mlFactoryMBean.getCacheHitsThroughput(), 0.0); assertEquals(cacheManager.mlFactoryMBean.getNumberOfCacheEvictions(), 0); } + + @Test + void verifyTimeBasedEviction() throws Exception { + ManagedLedgerFactoryConfig config = new ManagedLedgerFactoryConfig(); + config.setMaxCacheSize(1000); + config.setCacheEvictionFrequency(100); + config.setCacheEvictionTimeThresholdMillis(100); + + ManagedLedgerFactoryImpl factory = new ManagedLedgerFactoryImpl(bkc, bkc.getZkHandle(), config); + + ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("test"); + ManagedCursor c1 = ledger.openCursor("c1"); + c1.setActive(); + ManagedCursor c2 = ledger.openCursor("c2"); + c2.setActive(); + + EntryCacheManager cacheManager = factory.getEntryCacheManager(); + assertEquals(cacheManager.getSize(), 0); + + EntryCache cache = cacheManager.getEntryCache(ledger); + assertEquals(cache.getSize(), 0); + + ledger.addEntry(new byte[4]); + ledger.addEntry(new byte[3]); + + // Cache eviction should happen every 10 millis and clean all the entries older that 100ms + Thread.sleep(1000); + + c1.close(); + c2.close(); + + assertEquals(cacheManager.getSize(), 0); + assertEquals(cache.getSize(), 0); + + factory.shutdown(); + } + } 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 d0abbbbe3f295..35725e671bfcb 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 @@ -22,7 +22,6 @@ import static org.mockito.Matchers.anyInt; import static org.mockito.Mockito.doNothing; import static org.mockito.Mockito.doReturn; -import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; @@ -31,6 +30,13 @@ import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; +import com.google.common.base.Charsets; +import com.google.common.collect.Sets; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.PooledByteBufAllocator; +import io.netty.buffer.Unpooled; + import java.io.IOException; import java.lang.reflect.Field; import java.lang.reflect.Modifier; @@ -38,7 +44,6 @@ import java.security.GeneralSecurityException; import java.util.ArrayList; import java.util.Collections; -import java.util.EnumSet; import java.util.Iterator; import java.util.List; import java.util.Set; @@ -60,13 +65,10 @@ import org.apache.bookkeeper.client.BookKeeper; import org.apache.bookkeeper.client.BookKeeper.DigestType; import org.apache.bookkeeper.client.LedgerHandle; -import org.apache.bookkeeper.client.api.LedgerEntries; -import org.apache.bookkeeper.client.api.ReadHandle; import org.apache.bookkeeper.client.PulsarMockBookKeeper; import org.apache.bookkeeper.client.PulsarMockLedgerHandle; import org.apache.bookkeeper.client.api.LedgerEntries; import org.apache.bookkeeper.client.api.ReadHandle; -import org.apache.bookkeeper.client.api.WriteFlag; import org.apache.bookkeeper.conf.ClientConfiguration; import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.CloseCallback; @@ -86,6 +88,7 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerNotFoundException; import org.apache.bookkeeper.mledger.ManagedLedgerException.MetaStoreException; import org.apache.bookkeeper.mledger.ManagedLedgerFactory; +import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig; import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback; import org.apache.bookkeeper.mledger.impl.MetaStore.Stat; @@ -105,13 +108,6 @@ import org.slf4j.LoggerFactory; import org.testng.annotations.Test; -import com.google.common.base.Charsets; -import com.google.common.collect.Sets; - -import io.netty.buffer.ByteBuf; -import io.netty.buffer.PooledByteBufAllocator; -import io.netty.buffer.Unpooled; - public class ManagedLedgerTest extends MockedBookKeeperTestCase { private static final Logger log = LoggerFactory.getLogger(ManagedLedgerTest.class); @@ -1356,21 +1352,21 @@ public void invalidateConsumedEntriesFromCache() throws Exception { assertEquals(cacheManager.getSize(), entryCache.getSize()); c1.setReadPosition(p2); - ledger.discardEntriesFromCache(c1, p1); + ledger.discardEntriesFromCache(c1, p2); assertEquals(entryCache.getSize(), 7 * 3); assertEquals(cacheManager.getSize(), entryCache.getSize()); c1.setReadPosition(p3); - ledger.discardEntriesFromCache(c1, p2); - assertEquals(entryCache.getSize(), 7 * 2); + ledger.discardEntriesFromCache(c1, p3); + assertEquals(entryCache.getSize(), 7 * 3); assertEquals(cacheManager.getSize(), entryCache.getSize()); ledger.deactivateCursor(c1); - assertEquals(entryCache.getSize(), 7 * 2); // as c2.readPosition=p3 => Cache contains p3,p4 + assertEquals(entryCache.getSize(), 7 * 3); // as c2.readPosition=p3 => Cache contains p3,p4 assertEquals(cacheManager.getSize(), entryCache.getSize()); c2.setReadPosition(p4); - ledger.discardEntriesFromCache(c2, p3); + ledger.discardEntriesFromCache(c2, p4); assertEquals(entryCache.getSize(), 7); assertEquals(cacheManager.getSize(), entryCache.getSize()); @@ -1903,6 +1899,9 @@ public void testGetNextValidPosition() throws Exception { */ @Test public void testActiveDeactiveCursorWithDiscardEntriesFromCache() throws Exception { + ManagedLedgerFactoryConfig conf = new ManagedLedgerFactoryConfig(); + conf.setCacheEvictionFrequency(0.1); + ManagedLedgerFactory factory = new ManagedLedgerFactoryImpl(bkc, zkc, conf); ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("cache_eviction_ledger"); // Open Cursor also adds cursor into activeCursor-container @@ -1954,13 +1953,14 @@ public void testActiveDeactiveCursorWithDiscardEntriesFromCache() throws Excepti } // (3) Validate: cache should remove all entries read by both active cursors - log.info("expected, found : {}, {}", (5 * (totalInsertedEntries - readEntries)), entryCache.getSize()); - assertEquals((5 * (totalInsertedEntries - readEntries)), entryCache.getSize()); + log.info("expected, found : {}, {}", (5 * (totalInsertedEntries)), entryCache.getSize()); + assertEquals((5 * totalInsertedEntries), entryCache.getSize()); final int remainingEntries = totalInsertedEntries - readEntries; entries1 = cursor1.readEntries(remainingEntries); // Acknowledge only on last entry cursor1.markDelete(entries1.get(entries1.size() - 1).getPosition()); + for (Entry entry : entries1) { log.info("Read entry. Position={} Content='{}'", entry.getPosition(), new String(entry.getData())); entry.release(); @@ -1968,16 +1968,18 @@ public void testActiveDeactiveCursorWithDiscardEntriesFromCache() throws Excepti // (4) Validate: cursor2 is active cursor and has not read these entries yet: so, cache should not remove these // entries - assertEquals((5 * (remainingEntries)), entryCache.getSize()); + assertEquals((5 * totalInsertedEntries), entryCache.getSize()); + ledger.deactivateCursor(cursor1); ledger.deactivateCursor(cursor2); // (5) Validate: cursor2 is not active cursor now: cache should have removed all entries read by active cursor1 - assertEquals(0, entryCache.getSize()); + assertEquals(entryCache.getSize(), 0); log.info("Finished reading entries"); ledger.close(); + factory.shutdown(); } @Test @@ -2017,7 +2019,8 @@ public void testActiveDeactiveCursor() throws Exception { entry.release(); } - // (3) Validate: cache discards all entries as read by active cursor + // (3) Validate: cache discards all entries after all cursors are deactivated + ledger.deactivateCursor(cursor1); assertEquals(0, entryCache.getSize()); ledger.close(); @@ -2043,26 +2046,18 @@ public void testCursorRecoveryForEmptyLedgers() throws Exception { @Test public void testBacklogCursor() throws Exception { + int backloggedThreshold = 10; + ManagedLedgerFactoryConfig factoryConf = new ManagedLedgerFactoryConfig(); + factoryConf.setThresholdBackloggedCursor(backloggedThreshold); + ManagedLedgerFactory factory = new ManagedLedgerFactoryImpl(bkc, zkc, factoryConf); ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("cache_backlog_ledger"); - final long maxMessageCacheRetentionTimeMillis = 100; - Field field = ManagedLedgerImpl.class.getDeclaredField("maxMessageCacheRetentionTimeMillis"); - field.setAccessible(true); - Field modifiersField = Field.class.getDeclaredField("modifiers"); - modifiersField.setAccessible(true); - modifiersField.setInt(field, field.getModifiers() & ~Modifier.FINAL); - field.set(ledger, maxMessageCacheRetentionTimeMillis); - Field backlogThresholdField = ManagedLedgerImpl.class.getDeclaredField("maxActiveCursorBacklogEntries"); - backlogThresholdField.setAccessible(true); - final long maxActiveCursorBacklogEntries = (long) backlogThresholdField.get(ledger); - // Open Cursor also adds cursor into activeCursor-container ManagedCursor cursor1 = ledger.openCursor("c1"); ManagedCursor cursor2 = ledger.openCursor("c2"); - final int totalBacklogSizeEntries = (int) maxActiveCursorBacklogEntries; - CountDownLatch latch = new CountDownLatch(totalBacklogSizeEntries); - for (int i = 0; i < totalBacklogSizeEntries + 1; i++) { + CountDownLatch latch = new CountDownLatch(backloggedThreshold); + for (int i = 0; i < backloggedThreshold + 1; i++) { String content = "entry"; // 5 bytes ByteBuf entry = getMessageWithMetadata(content.getBytes()); ledger.asyncAddEntry(entry, new AddEntryCallback() { @@ -2086,12 +2081,8 @@ public void addFailed(ManagedLedgerException exception, Object ctx) { assertTrue(cursor1.isActive()); assertTrue(cursor2.isActive()); - // it allows message to be older enough to be considered in backlog - Thread.sleep(maxMessageCacheRetentionTimeMillis * 2); - // deactivate backlog cursors ledger.checkBackloggedCursors(); - Thread.sleep(100); // both cursors have to be inactive assertFalse(cursor1.isActive()); @@ -2114,6 +2105,8 @@ public void addFailed(ManagedLedgerException exception, Object ctx) { assertFalse(cursor2.isActive()); ledger.close(); + + factory.shutdown(); } @Test @@ -2240,12 +2233,12 @@ public void testManagedLedgerWithoutAutoCreate() throws Exception { assertFalse(factory.getManagedLedgers().containsKey("testManagedLedgerWithoutAutoCreate")); } - + @Test public void testManagedLedgerWithCreateLedgerTimeOut() throws Exception { ManagedLedgerConfig config = new ManagedLedgerConfig().setMetadataOperationsTimeoutSeconds(3); ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("timeout_ledger_test", config); - + BookKeeper bk = mock(BookKeeper.class); doNothing().when(bk).asyncCreateLedger(anyInt(), anyInt(), anyInt(), any(), any(), any(), any(), any()); AtomicInteger response = new AtomicInteger(0); @@ -2260,13 +2253,13 @@ public void createComplete(int rc, LedgerHandle lh, Object ctx) { latch.await(config.getMetadataOperationsTimeoutSeconds() + 2, TimeUnit.SECONDS); assertEquals(response.get(), BKException.Code.TimeoutException); - + ledger.close(); } - + /** * It verifies that asyncRead timesout if it doesn't receive response from bk-client in configured timeout - * + * * @throws Exception */ @Test @@ -2339,8 +2332,8 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) { /** * It verifies that if bk-client doesn't complete the add-entry in given time out then broker is resilient enought * to create new ledger and add entry successfully. - * - * + * + * * @throws Exception */ @Test(timeOut = 20000) @@ -2405,7 +2398,7 @@ private void setFieldValue(Class clazz, Object classObj, String fieldName, Objec field.setAccessible(true); field.set(classObj, fieldValue); } - + public static void retryStrategically(Predicate predicate, int retryCount, long intSleepTimeInMillis) throws Exception { for (int i = 0; i < retryCount; i++) { diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/util/RangeCacheTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/util/RangeCacheTest.java index d1d2e5d2c6305..27e63815c0107 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/util/RangeCacheTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/util/RangeCacheTest.java @@ -111,7 +111,7 @@ void simple() { @Test void customWeighter() { - RangeCache cache = new RangeCache<>(value -> value.s.length()); + RangeCache cache = new RangeCache<>(value -> value.s.length(), x -> 0); cache.put(0, new RefString("zero")); cache.put(1, new RefString("one")); @@ -120,6 +120,25 @@ void customWeighter() { assertEquals(cache.getNumberOfEntries(), 2); } + @Test + void customTimeExtraction() { + RangeCache cache = new RangeCache<>(value -> value.s.length(), x -> x.s.length()); + + cache.put(1, new RefString("1")); + cache.put(2, new RefString("22")); + cache.put(3, new RefString("333")); + cache.put(4, new RefString("4444")); + + assertEquals(cache.getSize(), 10); + assertEquals(cache.getNumberOfEntries(), 4); + + long evictedSize = cache.evictLEntriesBeforeTimestamp(3); + assertEquals(evictedSize, 6); + + assertEquals(cache.getSize(), 4); + assertEquals(cache.getNumberOfEntries(), 1); + } + @Test void doubleInsert() { RangeCache cache = new RangeCache<>(); @@ -172,7 +191,7 @@ void getRange() { @Test void eviction() { - RangeCache cache = new RangeCache<>(value -> value.s.length()); + RangeCache cache = new RangeCache<>(value -> value.s.length(), x -> 0); cache.put(0, new RefString("zero")); cache.put(1, new RefString("one")); 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 b600ac7618959..3825a86b45fc1 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 @@ -768,6 +768,16 @@ public class ServiceConfiguration implements PulsarConfiguration { doc = "Threshold to which bring down the cache level when eviction is triggered" ) private double managedLedgerCacheEvictionWatermark = 0.9f; + @FieldContext(category = CATEGORY_STORAGE_ML, + doc = "Configure the cache eviction frequency for the managed ledger cache. Default is 100/s") + private double managedLedgerCacheEvictionFrequency = 100.0; + @FieldContext(category = CATEGORY_STORAGE_ML, + doc = "All entries that have stayed in cache for more than the configured time, will be evicted") + private long managedLedgerCacheEvictionTimeThresholdMillis = 1000; + @FieldContext(category = CATEGORY_STORAGE_ML, + doc = "Configure the threshold (in number of entries) from where a cursor should be considered 'backlogged'" + + " and thus should be set as inactive.") + private long managedLedgerCursorBackloggedThreshold = 1000; @FieldContext( category = CATEGORY_STORAGE_ML, doc = "Rate limit the amount of writes per second generated by consumer acking the messages" diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java index 2b82204750ccb..e3a52d8d7d5fe 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java @@ -47,6 +47,9 @@ public ManagedLedgerClientFactory(ServiceConfiguration conf, ZooKeeper zkClient, managedLedgerFactoryConfig.setCacheEvictionWatermark(conf.getManagedLedgerCacheEvictionWatermark()); managedLedgerFactoryConfig.setNumManagedLedgerWorkerThreads(conf.getManagedLedgerNumWorkerThreads()); managedLedgerFactoryConfig.setNumManagedLedgerSchedulerThreads(conf.getManagedLedgerNumSchedulerThreads()); + managedLedgerFactoryConfig.setCacheEvictionFrequency(conf.getManagedLedgerCacheEvictionFrequency()); + managedLedgerFactoryConfig.setCacheEvictionTimeThresholdMillis(conf.getManagedLedgerCacheEvictionTimeThresholdMillis()); + managedLedgerFactoryConfig.setThresholdBackloggedCursor(conf.getManagedLedgerCursorBackloggedThreshold()); this.managedLedgerFactory = new ManagedLedgerFactoryImpl(bkClient, zkClient, managedLedgerFactoryConfig); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index 23ca67a87d170..ee4b05fb569f9 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -568,9 +568,12 @@ public CompletableFuture subscribe(final ServerCnx cnx, String subscri subscriptionFuture.thenAccept(subscription -> { try { + ledger.checkBackloggedCursors(); + Consumer consumer = new Consumer(subscription, subType, topic, consumerId, priorityLevel, consumerName, maxUnackedMessages, cnx, cnx.getRole(), metadata, readCompacted, initialPosition); subscription.addConsumer(consumer); + if (!cnx.isActive()) { consumer.close(); if (log.isDebugEnabled()) { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerRedeliveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerRedeliveryTest.java index fc05d986a7c3b..63c974dec5ee6 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerRedeliveryTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerRedeliveryTest.java @@ -36,6 +36,7 @@ public class ConsumerRedeliveryTest extends ProducerConsumerBase { @BeforeClass @Override protected void setup() throws Exception { + conf.setManagedLedgerCacheEvictionFrequency(0.1); super.internalSetup(); super.producerBaseSetup(); } @@ -62,7 +63,7 @@ public void testOrderedRedelivery() throws Exception { conf.setManagedLedgerMaxEntriesPerLedger(2); conf.setManagedLedgerMinLedgerRolloverTimeMinutes(0); - + ProducerBuilder producerBuilder = pulsarClient.newProducer().topic(topic) .producerName("my-producer-name"); Producer producer = producerBuilder.create(); @@ -99,7 +100,7 @@ public void testOrderedRedelivery() throws Exception { Message message = consumer1.receive(5, TimeUnit.SECONDS); MessageIdImpl msgId = (MessageIdImpl) message.getMessageId(); if (lastMsgId != null) { - assertTrue(lastMsgId.getLedgerId() <= msgId.getLedgerId()); + assertTrue(lastMsgId.getLedgerId() <= msgId.getLedgerId(), "lastMsgId: " + lastMsgId + " -- msgId: " + msgId); } lastMsgId = msgId; } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java index 74dd59d044b93..7192e7fa6dd6a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java @@ -796,9 +796,6 @@ public void testActiveAndInActiveConsumerEntryCacheBehavior() throws Exception { producer.send("message".getBytes()); msg = subscriber1.receive(5, TimeUnit.SECONDS); - // Verify: cache has to be cleared as there is no message needs to be consumed by active subscriber - assertEquals(entryCache.getSize(), 0, 1); - /************ usecase-2: *************/ // 1.b Subscriber slower-subscriber Consumer subscriber2 = pulsarClient.newConsumer() @@ -869,16 +866,8 @@ public void testDeactivatingBacklogConsumer() throws Exception { ManagedLedgerImpl ledger = (ManagedLedgerImpl) topicRef.getManagedLedger(); // reflection to set/get cache-backlog fields value: - final long maxMessageCacheRetentionTimeMillis = 100; - Field backlogThresholdField = ManagedLedgerImpl.class.getDeclaredField("maxActiveCursorBacklogEntries"); - backlogThresholdField.setAccessible(true); - Field field = ManagedLedgerImpl.class.getDeclaredField("maxMessageCacheRetentionTimeMillis"); - field.setAccessible(true); - Field modifiersField = Field.class.getDeclaredField("modifiers"); - modifiersField.setAccessible(true); - modifiersField.setInt(field, field.getModifiers() & ~Modifier.FINAL); - field.set(ledger, maxMessageCacheRetentionTimeMillis); - final long maxActiveCursorBacklogEntries = (long) backlogThresholdField.get(ledger); + final long maxMessageCacheRetentionTimeMillis = conf.getManagedLedgerCacheEvictionTimeThresholdMillis(); + final long maxActiveCursorBacklogEntries = conf.getManagedLedgerCursorBackloggedThreshold(); Message msg = null; final int totalMsgs = (int) maxActiveCursorBacklogEntries + receiverSize + 1; @@ -2911,16 +2900,16 @@ public void received(Consumer consumer, Message message) /** * This test verifies that broker activates fail-over consumer by considering priority-level as well. - * + * * 
      * 1. Start two failover consumer with same priority level, broker selects consumer based on name-sorting (consumer1).
      * 2. Switch non-active consumer to active (consumer2): by giving it higher priority
      * Partitioned-topic with 9 partitions:
      * 1. C1 (priority=1)
      * 2. C2,C3,C4 (priority=0)
-     * So, broker should evenly distribute C2,C3,C4 active consumers among 9 partitions. 
+     * So, broker should evenly distribute C2,C3,C4 active consumers among 9 partitions.
      *
- * + * * @throws Exception */ @Test diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerTest.java index 161c3742a8a06..fecef7946d5c1 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v1/V1_ProducerConsumerTest.java @@ -687,9 +687,6 @@ public void testActiveAndInActiveConsumerEntryCacheBehavior() throws Exception { producer.send("message".getBytes()); msg = subscriber1.receive(5, TimeUnit.SECONDS); - // Verify: cache has to be cleared as there is no message needs to be consumed by active subscriber - assertEquals(entryCache.getSize(), 0, 1); - /************ usecase-2: *************/ // 1.b Subscriber slower-subscriber Consumer subscriber2 = pulsarClient.newConsumer() @@ -732,93 +729,6 @@ public void testActiveAndInActiveConsumerEntryCacheBehavior() throws Exception { log.info("-- Exiting {} test --", methodName); } - @Test - public void testDeactivatingBacklogConsumer() throws Exception { - log.info("-- Starting {} test --", methodName); - - final long batchMessageDelayMs = 100; - final int receiverSize = 10; - final String topicName = "cache-topic"; - final String topic = "persistent://my-property/use/my-ns/" + topicName; - final String sub1 = "faster-sub1"; - final String sub2 = "slower-sub2"; - - // 1. Subscriber Faster subscriber: let it consume all messages immediately - Consumer subscriber1 = pulsarClient.newConsumer() - .topic("persistent://my-property/use/my-ns/" + topicName) - .subscriptionName(sub1) - .subscriptionType(SubscriptionType.Shared) - .receiverQueueSize(receiverSize) - .subscribe(); - // 1.b. Subscriber Slow subscriber: - Consumer subscriber2 = pulsarClient.newConsumer() - .topic("persistent://my-property/use/my-ns/" + topicName) - .subscriptionName(sub2) - .subscriptionType(SubscriptionType.Shared) - .receiverQueueSize(receiverSize) - .subscribe(); - Producer producer = pulsarClient.newProducer() - .topic(topic) - .enableBatching(batchMessageDelayMs != 0) - .batchingMaxPublishDelay(batchMessageDelayMs, TimeUnit.MILLISECONDS) - .batchingMaxMessages(5) - .create(); - - PersistentTopic topicRef = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topic).get(); - ManagedLedgerImpl ledger = (ManagedLedgerImpl) topicRef.getManagedLedger(); - - // reflection to set/get cache-backlog fields value: - final long maxMessageCacheRetentionTimeMillis = 100; - Field backlogThresholdField = ManagedLedgerImpl.class.getDeclaredField("maxActiveCursorBacklogEntries"); - backlogThresholdField.setAccessible(true); - Field field = ManagedLedgerImpl.class.getDeclaredField("maxMessageCacheRetentionTimeMillis"); - field.setAccessible(true); - Field modifiersField = Field.class.getDeclaredField("modifiers"); - modifiersField.setAccessible(true); - modifiersField.setInt(field, field.getModifiers() & ~Modifier.FINAL); - field.set(ledger, maxMessageCacheRetentionTimeMillis); - final long maxActiveCursorBacklogEntries = (long) backlogThresholdField.get(ledger); - - Messagemsg = null; - final int totalMsgs = (int) maxActiveCursorBacklogEntries + receiverSize + 1; - // 2. Produce messages - for (int i = 0; i < totalMsgs; i++) { - String message = "my-message-" + i; - producer.send(message.getBytes()); - } - // 3. Consume messages: at Faster subscriber - for (int i = 0; i < totalMsgs; i++) { - msg = subscriber1.receive(100, TimeUnit.MILLISECONDS); - subscriber1.acknowledge(msg); - } - - // wait : so message can be eligible to to be evict from cache - Thread.sleep(maxMessageCacheRetentionTimeMillis); - - // 4. deactivate subscriber which has built the backlog - ledger.checkBackloggedCursors(); - Thread.sleep(100); - - // 5. verify: active subscribers - Set activeSubscriber = Sets.newHashSet(); - ledger.getActiveCursors().forEach(c -> activeSubscriber.add(c.getName())); - assertTrue(activeSubscriber.contains(sub1)); - assertFalse(activeSubscriber.contains(sub2)); - - // 6. consume messages : at slower subscriber - for (int i = 0; i < totalMsgs; i++) { - msg = subscriber2.receive(100, TimeUnit.MILLISECONDS); - subscriber2.acknowledge(msg); - } - - ledger.checkBackloggedCursors(); - - activeSubscriber.clear(); - ledger.getActiveCursors().forEach(c -> activeSubscriber.add(c.getName())); - - assertTrue(activeSubscriber.contains(sub1)); - assertTrue(activeSubscriber.contains(sub2)); - } @Test(timeOut = 2000) public void testAsyncProducerAndConsumer() throws Exception { diff --git a/site2/docs/reference-configuration.md b/site2/docs/reference-configuration.md index 959f546128c1b..1f828adc17a12 100644 --- a/site2/docs/reference-configuration.md +++ b/site2/docs/reference-configuration.md @@ -176,6 +176,9 @@ Pulsar brokers are responsible for handling incoming messages from producers, di |managedLedgerDefaultAckQuorum| Number of guaranteed copies (acks to wait before write is complete) |2| |managedLedgerCacheSizeMB| Amount of memory to use for caching data payload in managed ledger. This memory is allocated from JVM direct memory and it’s shared across all the topics running in the same broker. By default, uses 1/5th of available direct memory || |managedLedgerCacheEvictionWatermark| Threshold to which bring down the cache level when eviction is triggered |0.9| +|managedLedgerCacheEvictionFrequency| Configure the cache eviction frequency for the managed ledger cache (evictions/sec) | 100.0 | +|managedLedgerCacheEvictionTimeThresholdMillis| All entries that have stayed in cache for more than the configured time, will be evicted | 1000 | +|managedLedgerCursorBackloggedThreshold| Configure the threshold (in number of entries) from where a cursor should be considered 'backlogged' and thus should be set as inactive. | 1000| |managedLedgerDefaultMarkDeleteRateLimit| Rate limit the amount of writes per second generated by consumer acking the messages |1.0| |managedLedgerMaxEntriesPerLedger| Max number of entries to append to a ledger before triggering a rollover. A ledger rollover is triggered on these conditions:
  • Either the max rollover time has been reached
  • or max entries have been written to the ledged and at least min-time has passed
|50000| |managedLedgerMinLedgerRolloverTimeMinutes| Minimum time between ledger rollover for a topic |10|