You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/07/24 08:11:28 UTC
(pulsar) branch branch-3.0 updated: [improve][test] Remove EntryCacheCreator from ManagedLedgerFactoryImpl (#24552)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new 05ea80ce973 [improve][test] Remove EntryCacheCreator from ManagedLedgerFactoryImpl (#24552)
05ea80ce973 is described below
commit 05ea80ce973f316942b7632e465b1a988717cae7
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Jul 24 08:24:31 2025 +0300
[improve][test] Remove EntryCacheCreator from ManagedLedgerFactoryImpl (#24552)
(cherry picked from commit 07af143f2282e1f9cc68cbaa32262093a4e29151)
---
.../mledger/impl/ManagedLedgerFactoryImpl.java | 30 ++++-------------
.../mledger/impl/cache/RangeEntryCacheImpl.java | 5 ++-
.../bookkeeper/mledger/impl/ManagedCursorTest.java | 38 ++++++++++++++++------
.../bookkeeper/test/MockedBookKeeperTestCase.java | 21 +-----------
4 files changed, 37 insertions(+), 57 deletions(-)
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 bef180eaf5d..fa0d8255207 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
@@ -21,7 +21,6 @@ package org.apache.bookkeeper.mledger.impl;
import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.bookkeeper.mledger.ManagedLedgerException.getManagedLedgerException;
import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
-import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Predicates;
import com.google.common.collect.Maps;
import io.netty.util.concurrent.DefaultThreadFactory;
@@ -42,7 +41,6 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import lombok.Getter;
@@ -150,7 +148,7 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
ManagedLedgerFactoryConfig config)
throws Exception {
this(metadataStore, new DefaultBkFactory(bkClientConfiguration),
- true /* isBookkeeperManaged */, config, NullStatsLogger.INSTANCE, null);
+ true /* isBookkeeperManaged */, config, NullStatsLogger.INSTANCE);
}
public ManagedLedgerFactoryImpl(MetadataStoreExtended metadataStore, BookKeeper bookKeeper)
@@ -164,21 +162,12 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
this(metadataStore, (policyConfig) -> CompletableFuture.completedFuture(bookKeeper), config);
}
- @VisibleForTesting
- public ManagedLedgerFactoryImpl(MetadataStoreExtended metadataStore, BookKeeper bookKeeper,
- Function<ManagedLedgerFactoryImpl, EntryCacheManager>
- entryCacheManagerCreator)
- throws Exception {
- this(metadataStore, __ -> CompletableFuture.completedFuture(bookKeeper), false,
- new ManagedLedgerFactoryConfig(), NullStatsLogger.INSTANCE,
- entryCacheManagerCreator);
- }
-
public ManagedLedgerFactoryImpl(MetadataStoreExtended metadataStore,
BookkeeperFactoryForCustomEnsemblePlacementPolicy bookKeeperGroupFactory,
ManagedLedgerFactoryConfig config)
throws Exception {
- this(metadataStore, bookKeeperGroupFactory, config, NullStatsLogger.INSTANCE);
+ this(metadataStore, bookKeeperGroupFactory, false /* isBookkeeperManaged */,
+ config, NullStatsLogger.INSTANCE);
}
public ManagedLedgerFactoryImpl(MetadataStoreExtended metadataStore,
@@ -186,16 +175,13 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
ManagedLedgerFactoryConfig config, StatsLogger statsLogger)
throws Exception {
this(metadataStore, bookKeeperGroupFactory, false /* isBookkeeperManaged */,
- config, statsLogger, null);
+ config, statsLogger);
}
private ManagedLedgerFactoryImpl(MetadataStoreExtended metadataStore,
BookkeeperFactoryForCustomEnsemblePlacementPolicy bookKeeperGroupFactory,
boolean isBookkeeperManaged,
- ManagedLedgerFactoryConfig config, StatsLogger statsLogger,
- Function<ManagedLedgerFactoryImpl, EntryCacheManager>
- entryCacheManagerCreator)
- throws Exception {
+ ManagedLedgerFactoryConfig config, StatsLogger statsLogger) throws Exception {
scheduledExecutor = OrderedScheduler.newSchedulerBuilder()
.numThreads(config.getNumManagedLedgerSchedulerThreads())
.statsLogger(statsLogger)
@@ -212,11 +198,7 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
config.getManagedCursorInfoCompressionType());
this.config = config;
this.mbean = new ManagedLedgerFactoryMBeanImpl(this);
- if (entryCacheManagerCreator == null) {
- this.entryCacheManager = new RangeEntryCacheManagerImpl(this, scheduledExecutor);
- } else {
- this.entryCacheManager = entryCacheManagerCreator.apply(this);
- }
+ this.entryCacheManager = new RangeEntryCacheManagerImpl(this, scheduledExecutor);
this.statsTask = scheduledExecutor.scheduleWithFixedDelay(catchingAndLoggingThrowables(this::refreshStats),
0, config.getStatsPeriodSeconds(), TimeUnit.SECONDS);
this.flushCursorsTask = scheduledExecutor.scheduleAtFixedRate(catchingAndLoggingThrowables(this::flushCursors),
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
index 26a958c6032..cfaad242d80 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
@@ -439,9 +439,8 @@ public class RangeEntryCacheImpl implements EntryCache {
* @param shouldCacheEntry if we should put the entry into the cache
* @return a handle to the operation
*/
- @VisibleForTesting
- protected CompletableFuture<List<EntryImpl>> readFromStorage(ReadHandle lh, long firstEntry, long lastEntry,
- boolean shouldCacheEntry) {
+ CompletableFuture<List<EntryImpl>> readFromStorage(ReadHandle lh,
+ long firstEntry, long lastEntry, boolean shouldCacheEntry) {
final int entriesToRead = (int) (lastEntry - firstEntry) + 1;
CompletableFuture<List<EntryImpl>> readResult = ReadEntryUtils.readAsync(ml, lh, firstEntry, lastEntry)
.thenApply(
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
index efe74a5330f..c3aa252e936 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
@@ -47,6 +47,7 @@ import java.util.BitSet;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -77,6 +78,8 @@ import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.LedgerEntry;
+import org.apache.bookkeeper.client.PulsarMockReadHandleInterceptor;
+import org.apache.bookkeeper.client.api.LedgerEntries;
import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback;
@@ -96,12 +99,11 @@ import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.ScanOutcome;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.VoidCallback;
import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback;
-import org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheImpl;
-import org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheManagerImpl;
import org.apache.bookkeeper.mledger.proto.MLDataFormats;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo;
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
+import org.apache.commons.collections4.iterators.EmptyIterator;
import org.apache.pulsar.common.api.proto.CommandSubscribe;
import org.apache.pulsar.common.api.proto.IntRange;
import org.apache.pulsar.common.util.FutureUtil;
@@ -134,7 +136,7 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
@AfterMethod
public void afterMethod() {
- setEntryCacheCreator(null);
+ bkc.setReadHandleInterceptor(null);
}
@Test
@@ -5272,16 +5274,32 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
@Test(timeOut = 10000)
public void testReadNoEntries() throws Exception {
final var firstRead = new AtomicBoolean(true);
- setEntryCacheCreator(ml -> new RangeEntryCacheImpl((RangeEntryCacheManagerImpl) factory.getEntryCacheManager(),
- ml, factory.getConfig().isCopyEntriesInCache()) {
-
+ bkc.setReadHandleInterceptor(new PulsarMockReadHandleInterceptor() {
@Override
- protected CompletableFuture<List<EntryImpl>> readFromStorage(ReadHandle lh, long firstEntry, long lastEntry,
- boolean shouldCacheEntry) {
+ public CompletableFuture<LedgerEntries> interceptReadAsync(long ledgerId, long firstEntry, long lastEntry,
+ LedgerEntries entries) {
if (firstRead.compareAndSet(true, false)) {
- return CompletableFuture.completedFuture(List.of());
+ // LedgerEntriesImpl doesn't allow empty entries list.
+ // Implementing a dummy LedgerEntries that returns an empty list.
+ return CompletableFuture.completedFuture(new LedgerEntries() {
+ @Override
+ public org.apache.bookkeeper.client.api.LedgerEntry getEntry(long entryId) {
+ return null;
+ }
+
+ @Override
+ public Iterator<org.apache.bookkeeper.client.api.LedgerEntry> iterator() {
+ return EmptyIterator.INSTANCE;
+ }
+
+ @Override
+ public void close() {
+
+ }
+ });
+ } else {
+ return CompletableFuture.completedFuture(entries);
}
- return super.readFromStorage(lh, firstEntry, lastEntry, shouldCacheEntry);
}
});
final var ml = factory.open("testReadNoEntries");
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 d30b4b84267..c7685cfaa65 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
@@ -23,16 +23,12 @@ import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
import lombok.SneakyThrows;
import org.apache.bookkeeper.client.PulsarMockBookKeeper;
import org.apache.bookkeeper.common.util.OrderedScheduler;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl;
-import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
-import org.apache.bookkeeper.mledger.impl.cache.EntryCache;
-import org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheManagerImpl;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
@@ -61,7 +57,6 @@ public abstract class MockedBookKeeperTestCase {
protected ExecutorService cachedExecutor;
protected FaultInjectionMetadataStore metadataStore;
- private Function<ManagedLedgerImpl, EntryCache> entryCacheCreator = null;
public MockedBookKeeperTestCase() {
// By default start a 3 bookies cluster
@@ -89,17 +84,7 @@ public abstract class MockedBookKeeperTestCase {
ManagedLedgerFactoryConfig managedLedgerFactoryConfig = new ManagedLedgerFactoryConfig();
initManagedLedgerFactoryConfig(managedLedgerFactoryConfig);
- factory = new ManagedLedgerFactoryImpl(metadataStore, bkc, __ -> new RangeEntryCacheManagerImpl(__,
- __.getScheduledExecutor()) {
-
- @Override
- public EntryCache getEntryCache(ManagedLedgerImpl ml) {
- if (entryCacheCreator != null) {
- return entryCacheCreator.apply(ml);
- }
- return super.getEntryCache(ml);
- }
- });
+ factory = new ManagedLedgerFactoryImpl(metadataStore, bkc);
setUpTestCase();
}
@@ -179,8 +164,4 @@ public abstract class MockedBookKeeperTestCase {
protected void stopMetadataStore() {
metadataStore.setAlwaysFail(new MetadataStoreException("failed"));
}
-
- protected void setEntryCacheCreator(Function<ManagedLedgerImpl, EntryCache> entryCacheCreator) {
- this.entryCacheCreator = entryCacheCreator;
- }
}