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 2026/09/16 13:26:44 UTC
(pulsar) branch lhotari-improve-ml-cache-adaptive created (now cd103880372c)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch lhotari-improve-ml-cache-adaptive
in repository https://gitbox.apache.org/repos/asf/pulsar.git
at cd103880372c [fix][ml] Clarify managed cache allocator default
This branch includes the following new commits:
new 002d49775025 [improve][ml] Use adaptive allocation for managed ledger cache copies
new cd103880372c [fix][ml] Clarify managed cache allocator default
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(pulsar) 01/02: [improve][ml] Use adaptive allocation for managed ledger cache copies
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lhotari-improve-ml-cache-adaptive
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 002d497750254069ab29e359ae1803520e061b99
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Wed Sep 16 16:08:56 2026 +0300
[improve][ml] Use adaptive allocation for managed ledger cache copies
Preserve explicit allocator overrides and cover shared batch buffer release. Fix mock batch size splitting after an empty entry and cover the BookKeeper batch-read kill switch.
Assisted-by: Codex
---
conf/pulsar_env.sh | 7 +-
.../mledger/impl/ManagedLedgerBatchReadTest.java | 99 ++++++++++++++++++++++
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 5 ++
.../impl/cache/RangeEntryCacheImplTest.java | 32 +++++--
.../common/allocator/PulsarByteBufAllocator.java | 15 +++-
.../allocator/PulsarByteBufAllocatorTest.java | 49 +++++++++++
.../bookkeeper/client/PulsarMockReadHandle.java | 2 +-
7 files changed, 194 insertions(+), 15 deletions(-)
diff --git a/conf/pulsar_env.sh b/conf/pulsar_env.sh
index 56709af34e25..8c2dae939dbc 100755
--- a/conf/pulsar_env.sh
+++ b/conf/pulsar_env.sh
@@ -51,10 +51,13 @@
# adaptive - Netty AdaptiveByteBufAllocator; auto-tunes pooling and prefers direct buffers.
# Named allocators override each setting independently with pulsar.allocator.<id>.<setting>:
# pulsar.allocator.default.type - allocator used by general Pulsar operations
-# pulsar.allocator.ml-cache.type - separate allocator used for managed-ledger cache copies
+# pulsar.allocator.ml-cache.type - separate allocator used for managed-ledger cache copies (default: adaptive)
# Supported settings: type, exit_on_oom (false), out_of_memory_policy (FallbackToHeap or ThrowException).
# Named settings fall back to the unqualified pulsar.allocator.<setting>, then the built-in default.
-# For example: -Dpulsar.allocator.type=pooled -Dpulsar.allocator.ml-cache.type=adaptive
+# Explicit global type and legacy pooled settings also apply to ml-cache unless overridden by name.
+# Batch reads copy entries into this cache even when managedLedgerCacheCopyEntries=false. Adaptive
+# reuses small size-class slots to limit fragmentation; retained chunks and size rounding still cost memory.
+# To retain the previous cache allocator: -Dpulsar.allocator.ml-cache.type=pooled
# Settings are read when an allocator is first created. default overrides do not apply to other IDs.
# Leak detection is global: use -Dio.netty.leakDetection.level=disabled|simple|advanced|paranoid.
# pulsar.allocator.leak_detection and per-allocator leak_detection settings are not supported.
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerBatchReadTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerBatchReadTest.java
new file mode 100644
index 000000000000..9432b74bf2a3
--- /dev/null
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerBatchReadTest.java
@@ -0,0 +1,99 @@
+/*
+ * 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.assertj.core.api.Assertions.assertThat;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class ManagedLedgerBatchReadTest extends MockedBookKeeperTestCase {
+ @DataProvider
+ public Object[][] readPaths() {
+ return new Object[][]{{false, false}, {false, true}, {true, false}, {true, true}};
+ }
+
+ @Test(dataProvider = "readPaths")
+ public void testBatchReadWithLeadingEmptyEntry(boolean reopen, boolean cacheEnabled) throws Exception {
+ factory.getEntryCacheManager().updateCacheSizeAndThreshold(cacheEnabled ? 1024 * 1024 : 0);
+ ManagedLedgerConfig config = new ManagedLedgerConfig().setEnsembleSize(1).setWriteQuorumSize(1)
+ .setAckQuorumSize(1);
+ config.setPulsarMessageEntries(false);
+ config.setBatchReadEnabled(true);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("batch-read", config);
+ ManagedCursor cursor = ledger.openCursor("cursor");
+ byte[][] payloads = {new byte[0], new byte[2048], new byte[]{42}};
+ for (byte[] payload : payloads) {
+ ledger.addEntry(payload);
+ }
+ // Keep the ledger's average entry size small enough that the cursor requests all three entries.
+ // The batch size limit must still split the unexpectedly large second entry into its own round.
+ for (int i = 0; i < 8; i++) {
+ ledger.addEntry(new byte[]{1});
+ }
+ if (reopen) {
+ ledger.close();
+ ledger = (ManagedLedgerImpl) factory.open("batch-read", config);
+ cursor = ledger.openCursor("cursor");
+ }
+ factory.getEntryCacheManager().clear();
+ List<List<Long>> rounds = new ArrayList<>();
+ bkc.setReadHandleInterceptor((ledgerId, first, last, entries) -> {
+ rounds.add(List.of(first, last));
+ return CompletableFuture.completedFuture(entries);
+ });
+ try {
+ CompletableFuture<List<Entry>> result = new CompletableFuture<>();
+ cursor.asyncReadEntries(3, 1024, new ReadEntriesCallback() {
+ @Override
+ public void readEntriesComplete(List<Entry> entries, Object ctx) {
+ result.complete(entries);
+ }
+
+ @Override
+ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
+ result.completeExceptionally(exception);
+ }
+ }, null, PositionFactory.LATEST);
+ List<Entry> entries = result.get(10, TimeUnit.SECONDS);
+ try {
+ assertThat(entries).hasSize(3);
+ for (int i = 0; i < payloads.length; i++) {
+ assertThat(entries.get(i).getData()).isEqualTo(payloads[i]);
+ }
+ assertThat(rounds).containsExactly(List.of(0L, 0L), List.of(1L, 1L), List.of(2L, 2L));
+ } finally {
+ entries.forEach(Entry::release);
+ }
+ } finally {
+ bkc.setReadHandleInterceptor(null);
+ ledger.close();
+ }
+ }
+}
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 16fb5d40a2dc..be3a6171947f 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
@@ -5969,6 +5969,11 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
ManagedLedgerImpl disabledLedger = (ManagedLedgerImpl) factory.open("batch_read_disabled", disabled);
assertFalse(disabledLedger.isBatchReadEnabled());
+ bkc.getConf().setBatchReadEnabled(false);
+ ManagedLedgerImpl disabledClientLedger = (ManagedLedgerImpl) factory.open("batch_read_client_disabled");
+ assertFalse(disabledClientLedger.isBatchReadEnabled());
+ bkc.getConf().setBatchReadEnabled(true);
+
bkc.getConf().setUseV2WireProtocol(false);
ManagedLedgerImpl v3ClientLedger = (ManagedLedgerImpl) factory.open("batch_read_v3_client");
assertFalse(v3ClientLedger.isBatchReadEnabled());
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java
index 8ec11a4c1a7d..dc0cdef0392a 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java
@@ -627,11 +627,14 @@ public class RangeEntryCacheImplTest {
LedgerHandle ledgerHandle = mock(LedgerHandle.class);
when(ledgerHandle.getId()).thenReturn(1L);
- // Create test entries for batch read
+ // Model BookKeeper batch decoding with slices of one large response buffer.
+ ByteBuf response = Unpooled.buffer(1024 * 1024);
+ response.writeZero(response.capacity());
List<LedgerEntry> entryList = new ArrayList<>();
for (long i = 0; i <= 4; i++) {
- entryList.add(LedgerEntryImpl.create(1L, i, 1, Unpooled.wrappedBuffer(new byte[]{(byte) i})));
+ entryList.add(LedgerEntryImpl.create(1L, i, 1, response.retainedSlice((int) i, 1)));
}
+ response.release();
LedgerEntries batchEntries = new LedgerEntries() {
@Override
public LedgerEntry getEntry(long entryId) {
@@ -661,15 +664,26 @@ public class RangeEntryCacheImplTest {
assertThat(future).isCompleted();
List<Entry> entries = future.getNow(null);
try {
- assertThat(entries).hasSize(5);
- for (int i = 0; i < 5; i++) {
- assertThat(entries.get(i).getEntryId()).isEqualTo(i);
+ try {
+ assertThat(entries).hasSize(5);
+ for (int i = 0; i < 5; i++) {
+ assertThat(entries.get(i).getEntryId()).isEqualTo(i);
+ }
+ // Verify batch read was used, not readUnconfirmedAsync
+ verify(ledgerHandle, never()).readUnconfirmedAsync(anyLong(), anyLong());
+ } finally {
+ entries.forEach(Entry::release);
+ }
+ cache.invalidateEntries(PositionFactory.create(1L, 4L));
+ assertThat(cache.getSize()).isEqualTo(1);
+ assertThat(response.refCnt()).as("cache must not retain the shared batch response").isZero();
+ ReferenceCountedEntry cached = cache.getEntries().get(PositionFactory.create(1L, 4L));
+ try {
+ assertThat(cached.getDataBuffer().getByte(0)).isZero();
+ } finally {
+ cached.release();
}
- // Verify batch read was used, not readUnconfirmedAsync
- verify(ledgerHandle, never()).readUnconfirmedAsync(anyLong(), anyLong());
} finally {
- entries.forEach(Entry::release);
- // Release the copies kept by the cache
cache.clear();
}
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/allocator/PulsarByteBufAllocator.java b/pulsar-common/src/main/java/org/apache/pulsar/common/allocator/PulsarByteBufAllocator.java
index a9b8a223ca5d..7f6cb458ac50 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/allocator/PulsarByteBufAllocator.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/allocator/PulsarByteBufAllocator.java
@@ -114,7 +114,7 @@ public class PulsarByteBufAllocator {
+ "." + key.substring(PROPERTY_PREFIX.length()));
return override != null ? override : propertyResolver.apply(key);
};
- ByteBufAllocator nettyAllocator = createNettyAllocator(resolveAllocatorType(resolver), name);
+ ByteBufAllocator nettyAllocator = createNettyAllocator(resolveAllocatorType(resolver, name), name);
return new RegisteredAllocator(createByteBufAllocator(name, resolver, nettyAllocator),
((ByteBufAllocatorMetricProvider) nettyAllocator).metric());
}).allocator();
@@ -131,12 +131,21 @@ public class PulsarByteBufAllocator {
@VisibleForTesting
static AllocatorType resolveAllocatorType(Function<String, String> propertyResolver) {
+ return resolveAllocatorType(propertyResolver, DEFAULT_ALLOCATOR_NAME);
+ }
+
+ private static AllocatorType resolveAllocatorType(Function<String, String> propertyResolver, String id) {
String type = propertyResolver.apply(PULSAR_ALLOCATOR_TYPE);
if (type != null) {
return AllocatorType.fromString(type);
}
String pooled = propertyResolver.apply(PULSAR_ALLOCATOR_POOLED);
- return pooled == null || "true".equalsIgnoreCase(pooled) ? AllocatorType.POOLED : AllocatorType.UNPOOLED;
+ if (pooled != null) {
+ return "true".equalsIgnoreCase(pooled) ? AllocatorType.POOLED : AllocatorType.UNPOOLED;
+ }
+ // Cache copies are commonly small, similarly sized entries. Adaptive reuses freed size-class slots
+ // within smaller chunks, limiting fragmentation from entries with different cache lifetimes.
+ return ML_CACHE_ALLOCATOR_NAME.equals(id) ? AllocatorType.ADAPTIVE : AllocatorType.POOLED;
}
private static ByteBufAllocator createNettyAllocator(AllocatorType type, String id) {
@@ -161,7 +170,7 @@ public class PulsarByteBufAllocator {
private static ByteBufAllocator createByteBufAllocator(String id, Function<String, String> propertyResolver,
ByteBufAllocator nettyAllocator) {
- final AllocatorType allocatorType = resolveAllocatorType(propertyResolver);
+ final AllocatorType allocatorType = resolveAllocatorType(propertyResolver, id);
final boolean isExitOnOutOfMemory = "true".equalsIgnoreCase(
propertyResolver.apply(PULSAR_ALLOCATOR_EXIT_ON_OOM));
final OutOfMemoryPolicy outOfMemoryPolicy = OutOfMemoryPolicy.valueOf(
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/allocator/PulsarByteBufAllocatorTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/allocator/PulsarByteBufAllocatorTest.java
index 21715549443a..809962b4190d 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/allocator/PulsarByteBufAllocatorTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/allocator/PulsarByteBufAllocatorTest.java
@@ -27,6 +27,7 @@ import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.ByteBufAllocatorMetric;
import io.netty.buffer.PooledByteBufAllocatorMetric;
+import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.util.ResourceLeakDetector;
import java.util.ArrayList;
import java.util.List;
@@ -83,6 +84,54 @@ public class PulsarByteBufAllocatorTest {
}
}
+ @DataProvider
+ public Object[][] cacheAllocatorDefaults() {
+ return new Object[][]{
+ {null, null, null, AdaptiveByteBufAllocator.class},
+ {"pooled", null, null, PooledByteBufAllocatorMetric.class},
+ {"unpooled", null, null, UnpooledByteBufAllocator.DEFAULT.metric().getClass()},
+ {null, "true", null, PooledByteBufAllocatorMetric.class},
+ {null, "false", null, UnpooledByteBufAllocator.DEFAULT.metric().getClass()},
+ {"adaptive", "false", null, AdaptiveByteBufAllocator.class},
+ {"pooled", null, "adaptive", AdaptiveByteBufAllocator.class},
+ {"adaptive", null, "pooled", PooledByteBufAllocatorMetric.class},
+ {null, null, "unpooled", UnpooledByteBufAllocator.DEFAULT.metric().getClass()}
+ };
+ }
+
+ @Test(dataProvider = "cacheAllocatorDefaults")
+ public void testCacheAllocatorDefaultAndOverrides(String globalType, String legacyPooled, String cacheType,
+ Class<?> expectedMetricType) {
+ Properties properties = new Properties();
+ if (globalType != null) {
+ properties.setProperty("pulsar.allocator.type", globalType);
+ }
+ if (legacyPooled != null) {
+ properties.setProperty("pulsar.allocator.pooled", legacyPooled);
+ }
+ if (cacheType != null) {
+ properties.setProperty("pulsar.allocator.ml-cache.type", cacheType);
+ }
+ AllocatorRegistry registry = new AllocatorRegistry(properties::getProperty);
+ ByteBufAllocator cache = registry.getOrCreate(ML_CACHE_ALLOCATOR_NAME);
+ assertThat(registry.getAllocatorMetric(ML_CACHE_ALLOCATOR_NAME)).isInstanceOf(expectedMetricType);
+ ByteBuf buffer = cache.directBuffer(100, 100);
+ try {
+ buffer.writeByte(42);
+ assertThat(buffer.isDirect()).isTrue();
+ assertThat(buffer.readByte()).isEqualTo((byte) 42);
+ } finally {
+ buffer.release();
+ }
+ if (globalType == null && legacyPooled == null) {
+ registry.getOrCreate(DEFAULT_ALLOCATOR_NAME);
+ registry.getOrCreate("other");
+ assertThat(registry.getAllocatorMetric(DEFAULT_ALLOCATOR_NAME))
+ .isInstanceOf(PooledByteBufAllocatorMetric.class);
+ assertThat(registry.getAllocatorMetric("other")).isInstanceOf(PooledByteBufAllocatorMetric.class);
+ }
+ }
+
@Test
public void testNamedOverridesAndGlobalFallback() {
Properties properties = new Properties();
diff --git a/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockReadHandle.java b/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockReadHandle.java
index c1d81456a64c..c8e3d6b41d44 100644
--- a/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockReadHandle.java
+++ b/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockReadHandle.java
@@ -115,7 +115,7 @@ class PulsarMockReadHandle extends LedgerHandle {
long lastEntry = firstEntry;
for (long eid = firstEntry; eid <= lastEntryByCount; eid++) {
long entrySize = entries.get((int) eid).getLength();
- if (maxSize > 0 && accumulatedSize > 0 && accumulatedSize + entrySize > maxSize) {
+ if (maxSize > 0 && eid > firstEntry && accumulatedSize + entrySize > maxSize) {
break;
}
accumulatedSize += entrySize;
(pulsar) 02/02: [fix][ml] Clarify managed cache allocator default
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lhotari-improve-ml-cache-adaptive
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit cd103880372c6498c5fcc19f992a5137fd874de1
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Wed Sep 16 16:26:31 2026 +0300
[fix][ml] Clarify managed cache allocator default
Keep the pooled fallback explicit for the default allocator while using adaptive for ml-cache when no allocator type is configured.
Assisted-by: Codex
---
.../org/apache/pulsar/common/allocator/PulsarByteBufAllocator.java | 3 +++
1 file changed, 3 insertions(+)
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/allocator/PulsarByteBufAllocator.java b/pulsar-common/src/main/java/org/apache/pulsar/common/allocator/PulsarByteBufAllocator.java
index 7f6cb458ac50..ecee291778d6 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/allocator/PulsarByteBufAllocator.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/allocator/PulsarByteBufAllocator.java
@@ -143,6 +143,9 @@ public class PulsarByteBufAllocator {
if (pooled != null) {
return "true".equalsIgnoreCase(pooled) ? AllocatorType.POOLED : AllocatorType.UNPOOLED;
}
+ if (DEFAULT_ALLOCATOR_NAME.equals(id)) {
+ return AllocatorType.POOLED;
+ }
// Cache copies are commonly small, similarly sized entries. Adaptive reuses freed size-class slots
// within smaller chunks, limiting fragmentation from entries with different cache lifetimes.
return ML_CACHE_ALLOCATOR_NAME.equals(id) ? AllocatorType.ADAPTIVE : AllocatorType.POOLED;