Merge remote-tracking branch 'apache/master' into mmerli/ml-ledger-ordering-key

# Conflicts:
#	managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java

Author: merlimat

CONTRIBUTING.md

@@ -283,56 +283,11 @@ disallows privileged containers (the run continues either way, with less accurat
 `-Pdocker.wolfi` builds the base image from Wolfi, which is what makes the `GLIBC_TUNABLES` the test
 sets take effect.
 
-#### Analyzing a JFR file
+#### Performance recording analysis
 
-A recording collects several events at once, and each of them is worth looking at from more than one
-angle, so **`jfrFlamegraphs`** renders the lot in one go:
-
-```bash
-./gradlew jfrFlamegraphs                              # every .jfr in build/test-profiles
-./gradlew jfrFlamegraphs -Pjfr=tests/integration/build # the integration-test recordings
-./gradlew jfrFlamegraphs -Pjfr=<some-file>.jfr         # just one recording
-```
-
-Each recording gets a directory beside it named after the file without its extension plus a
-`-flamegraphs` suffix — `profile.jfr` produces `profile-flamegraphs/` — holding one flame graph per
-view: `cpu`, `wall`, `alloc` and `lock`, each rendered merged (`cpu.html`), split per thread
-(`cpu_threads.html`) and grouped into async-profiler's categories (`cpu_classify.html`). A view whose
-event the recording does not contain is skipped rather than failing the run. Pass
-`-Pjfr.types=cpu,nativemem` to render a different set of events, and `-Pjfrconv=<path>` when
-async-profiler's `jfrconv` is neither next to the library `LIBASYNCPROFILER_PATH` points at nor in
-the JDK that runs Gradle (Amazon Corretto ships it).
-
-For anything the flame graphs don't answer, open the `.jfr` itself in
-[Eclipse Mission Control](https://adoptium.net/jmc) or IntelliJ IDEA, or run `jfrconv` by hand.
-
-> Do not judge a recording by `jfr summary` (the JDK's own tool). The default options record through
-> `jfrsync`, and `jfr summary` reports only a handful of `jdk.ExecutionSample` events for a recording
-> that `jfrconv` reads hundreds of profiler samples from — it looks empty when it is not.
-
-#### Agent-assisted analysis with the Jafar MCP server
-
-The [Jafar MCP server](https://github.com/btraceio/jafar/blob/main/jfr-mcp/README.md) lets an AI
-coding agent read a JFR recording directly, which turns a flame graph into something you can ask
-questions about. Register it once (it needs [JBang](https://www.jbang.dev/) and JDK 25+):
-
-```bash
-claude mcp add jafar -- jbang jfr-mcp@btraceio --stdio
-```
-
-It exposes `jfr_diagnose` (automated diagnosis of a recording), `jfr_stackprofile` (structured stack
-profiling with a time-series and per-thread breakdown), `jfr_hotmethods`, `jfr_flamegraph`,
-`jfr_callgraph`, `jfr_exceptions`, `jfr_tsa` (thread-state analysis), `jfr_use` (USE method) and
-`jfr_query` for [JfrPath](https://github.com/btraceio/jafar) queries. A prompt that works well as a
-starting point:
-
-> use Jafar MCP's jfr_diagnose and jfr_stackprofile to analyze @filename.jfr. Besides showing the
-> report on the console, write the analysis in a markdown file with the jfr file as prefix and the
-> suffix as ".analysis.md"
-
-Treat the result as a lead to verify, not a conclusion: the
-[agent guardrails](AGENTS.md) apply here as much as anywhere, and a performance claim still needs a
-benchmark or a second profile behind it.
+See [`tests/performance/README.md`](tests/performance/README.md) for JFR rendering, async-profiler
+recording analysis, Jafar MCP usage and MAT MCP based memory-leak investigation. That document also
+explains how to keep the raw recordings and analysis next to the workload documentation.
 
 ### Integration tests
 

conf/broker.conf

@@ -1481,6 +1481,18 @@ managedLedgerCacheEvictionExtendTTLOfEntriesWithRemainingExpectedReadsMaxTimes=5
 # Default is true, to behave like a LRU cache.
 managedLedgerCacheEvictionExtendTTLOfRecentlyAccessed=true
 
+# Enable the BookKeeper batch read API for managed-ledger storage reads: a single RPC fetches
+# multiple entries, reducing network overhead for sequential reads. Batch read requires
+# bookkeeperUseV2WireProtocol and BookKeeper's own batch read flag (bookkeeper_batchReadEnabled),
+# checked on the BookKeeper client when a topic is loaded: regular reads are used otherwise, as
+# well as for striped ledgers (managedLedgerDefaultEnsembleSize differs from
+# managedLedgerDefaultWriteQuorum) and for bookies without batch read support. Each batch read
+# request is bounded by the size limit of the dispatcher read that triggered it (e.g.
+# dispatcherMaxReadSizeBytes) and by the BookKeeper client's max frame size (maxMessageSize plus
+# padding); a read needing more data is split into sequential batch read requests. Entries read
+# this way are copied when inserted in the entry cache.
+managedLedgerBatchReadEnabled=true
+
 # This setting configures the duration of continuing to cache added entries while there are no
 # active cursors, when the last active cursor has left or immediately after initialization when
 # the persistent topic and the managed ledger gets loaded.

conf/pulsar_env.sh

@@ -49,6 +49,18 @@
 #   pooled   - Netty PooledByteBufAllocator; prefers direct buffers (default).
 #   unpooled - Netty UnpooledByteBufAllocator; prefers heap buffers.
 #   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 (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.
+# 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.
 # -Dpulsar.allocator.pooled=true is deprecated; use -Dpulsar.allocator.type=pooled instead.
 # pulsar.allocator.type takes precedence over the legacy pulsar.allocator.pooled property.
 # If pulsar.allocator.type is unset, pulsar.allocator.pooled=true (or unset) selects pooled;

conf/standalone.conf

@@ -981,6 +981,18 @@ managedLedgerCacheEvictionExtendTTLOfEntriesWithRemainingExpectedReadsMaxTimes=5
 # Default is true, to behave like a LRU cache.
 managedLedgerCacheEvictionExtendTTLOfRecentlyAccessed=true
 
+# Enable the BookKeeper batch read API for managed-ledger storage reads: a single RPC fetches
+# multiple entries, reducing network overhead for sequential reads. Batch read requires
+# bookkeeperUseV2WireProtocol and BookKeeper's own batch read flag (bookkeeper_batchReadEnabled),
+# checked on the BookKeeper client when a topic is loaded: regular reads are used otherwise, as
+# well as for striped ledgers (managedLedgerDefaultEnsembleSize differs from
+# managedLedgerDefaultWriteQuorum) and for bookies without batch read support. Each batch read
+# request is bounded by the size limit of the dispatcher read that triggered it (e.g.
+# dispatcherMaxReadSizeBytes) and by the BookKeeper client's max frame size (maxMessageSize plus
+# padding); a read needing more data is split into sequential batch read requests. Entries read
+# this way are copied when inserted in the entry cache.
+managedLedgerBatchReadEnabled=true
+
 # This setting configures the duration of continuing to cache added entries while there are no
 # active cursors, when the last active cursor has left or immediately after initialization when
 # the persistent topic and the managed ledger gets loaded.

managed-ledger/src/main/java/org/apache/bookkeeper/client/BookKeeperClientConfigAccessor.java

@@ -0,0 +1,40 @@
+/*
+ * 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.client;
+
+import org.apache.bookkeeper.conf.ClientConfiguration;
+
+/**
+ * Gives the managed ledger access to what a {@link BookKeeper} client's configuration says about the client, whose
+ * accessor is protected: this class lives in the client's package for that purpose.
+ */
+public final class BookKeeperClientConfigAccessor {
+
+    private BookKeeperClientConfigAccessor() {
+    }
+
+    /**
+     * Whether the client can issue batch reads: they require the v2 wire protocol and the client's own batch read
+     * flag. A client without a configuration (a mock) cannot.
+     */
+    public static boolean supportsBatchRead(BookKeeper bookKeeper) {
+        ClientConfiguration conf = bookKeeper.getConf();
+        return conf != null && conf.getUseV2WireProtocol() && conf.isBatchReadEnabled();
+    }
+}

managed-ledger/src/main/java/org/apache/bookkeeper/client/package-info.java

@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+/**
+ * Managed ledger access to BookKeeper client internals that the client does not expose outside its package.
+ */
+package org.apache.bookkeeper.client;

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java

@@ -88,6 +88,20 @@ public class ManagedLedgerConfig {
     @Getter
     @Setter
     private boolean cacheEvictionByExpectedReadCount = true;
+
+    /**
+     * Enable the BookKeeper batch read API when reading entries from bookkeeper: a single RPC fetches multiple
+     * entries, reducing network overhead. It is only used when the BookKeeper client supports it (v2 wire protocol
+     * with batch reads enabled), which the managed ledger checks when it is opened; the client uses regular reads for
+     * striped ledgers (ensembleSize differs from writeQuorumSize) and for bookies without batch read support. Each
+     * batch read request is bounded by the size limit of the read that triggered it and by the client's max frame
+     * size, a read needing more data being split into sequential requests. Entries read this way are copied when
+     * inserted in the entry cache, since their buffers are slices of a shared response frame.
+     */
+    @Getter
+    @Setter
+    private boolean batchReadEnabled = true;
+
     /**
      * Whether the entries of this managed ledger are Pulsar messages, so that an entry's payload begins with the
      * headers that parse into a {@code MessageMetadata}.

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/OpenTelemetryManagedLedgerCacheStats.java

@@ -18,12 +18,13 @@
  */
 package org.apache.bookkeeper.mledger;
 
+import static org.apache.pulsar.common.allocator.PulsarByteBufAllocator.ML_CACHE_ALLOCATOR_NAME;
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.api.metrics.BatchCallback;
 import io.opentelemetry.api.metrics.ObservableLongMeasurement;
 import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl;
-import org.apache.bookkeeper.mledger.impl.cache.PooledByteBufAllocatorStats;
-import org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheImpl;
+import org.apache.pulsar.common.allocator.ByteBufAllocatorStats;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
 import org.apache.pulsar.opentelemetry.Constants;
 import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes.CacheEntryStatus;
 import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes.CacheOperationStatus;
@@ -160,7 +161,8 @@ private void recordMetrics(ManagedLedgerFactoryImpl factory) {
         cacheOperationCounter.record(stats.getCacheMissesTotal(), CacheOperationStatus.MISS.attributes);
         cacheOperationBytesCounter.record(stats.getCacheMissesBytesTotal(), CacheOperationStatus.MISS.attributes);
 
-        var allocatorStats = new PooledByteBufAllocatorStats(RangeEntryCacheImpl.ALLOCATOR);
+        var allocatorStats = new ByteBufAllocatorStats(
+                PulsarByteBufAllocator.getAllocatorMetric(ML_CACHE_ALLOCATOR_NAME));
         cachePoolActiveAllocationCounter.record(allocatorStats.activeAllocationsSmall, PoolArenaType.SMALL.attributes);
         cachePoolActiveAllocationCounter.record(allocatorStats.activeAllocationsNormal,
                 PoolArenaType.NORMAL.attributes);

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java

@@ -960,12 +960,20 @@ public void asyncReadEntriesWithSkip(int numberOfEntriesToRead, long maxSizeByte
         }
 
         int numOfEntriesToRead = applyMaxSizeCap(numberOfEntriesToRead, maxSizeBytes);
+        readEntriesWithSkip(numOfEntriesToRead, maxSizeBytes, callback, ctx, maxPosition, skipCondition);
+    }
 
+    /**
+     * Reads {@code numOfEntriesToRead} entries, a count that the caller already capped with {@code maxSizeBytes}, the
+     * size limit the read carries along to bound its storage requests.
+     */
+    private void readEntriesWithSkip(int numOfEntriesToRead, long maxSizeBytes, ReadEntriesCallback callback,
+                                     Object ctx, Position maxPosition, Predicate<Position> skipCondition) {
         PENDING_READ_OPS_UPDATER.incrementAndGet(this);
         // Skip deleted entries.
         skipCondition = skipCondition == null ? this::isMessageDeleted : skipCondition.or(this::isMessageDeleted);
-        OpReadEntry op =
-            OpReadEntry.create(this, readPosition, numOfEntriesToRead, callback, ctx, maxPosition, skipCondition, true);
+        OpReadEntry op = OpReadEntry.create(this, readPosition, numOfEntriesToRead, maxSizeBytes, callback, ctx,
+                maxPosition, skipCondition, true);
         ledger.asyncReadEntries(op);
     }
 
@@ -1117,12 +1125,11 @@ public void asyncReadEntriesWithSkipOrWait(int maxEntries, long maxSizeBytes, Re
         if (hasMoreEntries() && maxPosition.compareTo(readPosition) >= 0) {
             // If we have available entries, we can read them immediately
             log.debug("Read entries immediately");
-            asyncReadEntriesWithSkip(numberOfEntriesToRead, NO_MAX_SIZE_LIMIT, callback, ctx,
-                    maxPosition, skipCondition);
+            readEntriesWithSkip(numberOfEntriesToRead, maxSizeBytes, callback, ctx, maxPosition, skipCondition);
         } else {
             // Skip deleted entries.
             skipCondition = skipCondition == null ? this::isMessageDeleted : skipCondition.or(this::isMessageDeleted);
-            OpReadEntry op = OpReadEntry.create(this, readPosition, numberOfEntriesToRead, callback,
+            OpReadEntry op = OpReadEntry.create(this, readPosition, numberOfEntriesToRead, maxSizeBytes, callback,
                     ctx, maxPosition, skipCondition, true);
             int opReadId = op.id;
             if (!WAITING_READ_OP_UPDATER.compareAndSet(this, null, op)) {

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

@@ -76,6 +76,7 @@
 import org.apache.bookkeeper.client.BKException.Code;
 import org.apache.bookkeeper.client.BookKeeper;
 import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.client.BookKeeperClientConfigAccessor;
 import org.apache.bookkeeper.client.LedgerHandle;
 import org.apache.bookkeeper.client.api.LedgerEntry;
 import org.apache.bookkeeper.client.api.LedgerMetadata;
@@ -161,6 +162,8 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
     private static final Logger slog = Logger.get(ManagedLedgerImpl.class);
 
     protected final BookKeeper bookKeeper;
+    /** Whether the BookKeeper client can batch read: the v2 wire protocol with batch reads enabled. */
+    private final boolean batchReadSupported;
     protected final String name;
     protected final Logger log;
 
@@ -386,6 +389,7 @@ public ManagedLedgerImpl(ManagedLedgerFactoryImpl factory, BookKeeper bookKeeper
         }
         this.factory = factory;
         this.bookKeeper = bookKeeper;
+        this.batchReadSupported = BookKeeperClientConfigAccessor.supportsBatchRead(bookKeeper);
         this.config = config;
         this.store = store;
         this.name = name;
@@ -1934,6 +1938,14 @@ synchronized void addEntryFailedDueToConcurrentlyModified(final LedgerHandle cur
                                 + " position when the ledger was concurrently modified"
                                 + " (the ledger may be closed by auto-replication)");
                 ledgerClosed(currentLedger, lh.getLastAddConfirmed());
+                // Close the abandoned write handle, or it leaks with its periodic explicit-LAC flush task.
+                currentLedger.asyncClose((closeRc, closedLedger, closeCtx) -> {
+                    if (closeRc != Code.OK) {
+                        log.debug().attr("ledgerId", currentLedger.getId())
+                                .attr("status", BKException.getMessage(closeRc))
+                                .log("Error when closing ledger after it was concurrently modified");
+                    }
+                }, null);
             } else {
                 log.error().attr("ledgerId", currentLedger.getId())
                     .attr("lastAddConfirmed", currentLedger.getLastAddConfirmed())
@@ -2492,12 +2504,14 @@ protected void asyncReadEntry(ReadHandle ledger, long firstEntry, long lastEntry
         if (config.getReadEntryTimeoutSeconds() > 0) {
             ReadEntryCallbackWrapper readCallback = ReadEntryCallbackWrapper.create(this, ledger.getId(), firstEntry,
                     opReadEntry, ctx, timeoutAtNanos(config.getReadEntryTimeoutSeconds()));
-            entryCache.asyncReadEntry(ledger, firstEntry, lastEntry, expectedReadCount, readCallback, ctx);
+            entryCache.asyncReadEntry(ledger, firstEntry, lastEntry, opReadEntry.maxSizeBytes, expectedReadCount,
+                    readCallback, ctx);
             if (readCallback.registerTimeout()) {
                 factory.getReadEntryTimeoutTracker().add(readCallback);
             }
         } else {
-            entryCache.asyncReadEntry(ledger, firstEntry, lastEntry, expectedReadCount, opReadEntry, ctx);
+            entryCache.asyncReadEntry(ledger, firstEntry, lastEntry, opReadEntry.maxSizeBytes, expectedReadCount,
+                    opReadEntry, ctx);
         }
     }
 
@@ -4562,6 +4576,14 @@ public ManagedLedgerConfig getConfig() {
         return config;
     }
 
+    /**
+     * Whether storage reads use the BookKeeper batch read API: it must be enabled in the config and supported by
+     * the BookKeeper client (v2 wire protocol with batch reads enabled in its configuration).
+     */
+    public boolean isBatchReadEnabled() {
+        return batchReadSupported && config.isBatchReadEnabled();
+    }
+
     @Override
     public void setConfig(ManagedLedgerConfig config) {
         this.config = config;