lh fix individual read deduplication and limiting

distribution/server/src/assemble/LICENSE.bin.txt

@@ -338,12 +338,11 @@ The Apache Software License, Version 2.0
     - io.prometheus-simpleclient_tracer_otel-0.16.0.jar
     - io.prometheus-simpleclient_tracer_otel_agent-0.16.0.jar
  * Prometheus exporter
-    - io.prometheus-prometheus-metrics-config-1.3.1.jar
-    - io.prometheus-prometheus-metrics-exporter-common-1.3.1.jar
-    - io.prometheus-prometheus-metrics-exporter-httpserver-1.3.1.jar
-    - io.prometheus-prometheus-metrics-exposition-formats-1.3.1.jar
-    - io.prometheus-prometheus-metrics-model-1.3.1.jar
-    - io.prometheus-prometheus-metrics-shaded-protobuf-1.3.1.jar
+    - io.prometheus-prometheus-metrics-config-1.3.3.jar
+    - io.prometheus-prometheus-metrics-exporter-common-1.3.3.jar
+    - io.prometheus-prometheus-metrics-exporter-httpserver-1.3.3.jar
+    - io.prometheus-prometheus-metrics-exposition-formats-1.3.3.jar
+    - io.prometheus-prometheus-metrics-model-1.3.3.jar
  * Jakarta Bean Validation API
     - jakarta.validation-jakarta.validation-api-2.0.2.jar
     - javax.validation-validation-api-1.1.0.Final.jar
@@ -516,21 +515,21 @@ The Apache Software License, Version 2.0
   * RoaringBitmap
     - org.roaringbitmap-RoaringBitmap-1.2.0.jar
   * OpenTelemetry
-    - io.opentelemetry-opentelemetry-api-1.41.0.jar
-    - io.opentelemetry-opentelemetry-api-incubator-1.41.0-alpha.jar
-    - io.opentelemetry-opentelemetry-context-1.41.0.jar
-    - io.opentelemetry-opentelemetry-exporter-common-1.41.0.jar
-    - io.opentelemetry-opentelemetry-exporter-otlp-1.41.0.jar
-    - io.opentelemetry-opentelemetry-exporter-otlp-common-1.41.0.jar
-    - io.opentelemetry-opentelemetry-exporter-prometheus-1.41.0-alpha.jar
-    - io.opentelemetry-opentelemetry-exporter-sender-okhttp-1.41.0.jar
-    - io.opentelemetry-opentelemetry-sdk-1.41.0.jar
-    - io.opentelemetry-opentelemetry-sdk-common-1.41.0.jar
-    - io.opentelemetry-opentelemetry-sdk-extension-autoconfigure-1.41.0.jar
-    - io.opentelemetry-opentelemetry-sdk-extension-autoconfigure-spi-1.41.0.jar
-    - io.opentelemetry-opentelemetry-sdk-logs-1.41.0.jar
-    - io.opentelemetry-opentelemetry-sdk-metrics-1.41.0.jar
-    - io.opentelemetry-opentelemetry-sdk-trace-1.41.0.jar
+    - io.opentelemetry-opentelemetry-api-1.44.1.jar
+    - io.opentelemetry-opentelemetry-api-incubator-1.44.1-alpha.jar
+    - io.opentelemetry-opentelemetry-context-1.44.1.jar
+    - io.opentelemetry-opentelemetry-exporter-common-1.44.1.jar
+    - io.opentelemetry-opentelemetry-exporter-otlp-1.44.1.jar
+    - io.opentelemetry-opentelemetry-exporter-otlp-common-1.44.1.jar
+    - io.opentelemetry-opentelemetry-exporter-prometheus-1.44.1-alpha.jar
+    - io.opentelemetry-opentelemetry-exporter-sender-okhttp-1.44.1.jar
+    - io.opentelemetry-opentelemetry-sdk-1.44.1.jar
+    - io.opentelemetry-opentelemetry-sdk-common-1.44.1.jar
+    - io.opentelemetry-opentelemetry-sdk-extension-autoconfigure-1.44.1.jar
+    - io.opentelemetry-opentelemetry-sdk-extension-autoconfigure-spi-1.44.1.jar
+    - io.opentelemetry-opentelemetry-sdk-logs-1.44.1.jar
+    - io.opentelemetry-opentelemetry-sdk-metrics-1.44.1.jar
+    - io.opentelemetry-opentelemetry-sdk-trace-1.44.1.jar
     - io.opentelemetry.instrumentation-opentelemetry-instrumentation-api-1.33.6.jar
     - io.opentelemetry.instrumentation-opentelemetry-instrumentation-api-semconv-1.33.6-alpha.jar
     - io.opentelemetry.instrumentation-opentelemetry-resources-1.33.6-alpha.jar
@@ -572,10 +571,8 @@ Protocol Buffers License
 CDDL-1.1 -- ../licenses/LICENSE-CDDL-1.1.txt
  * Java Annotations API
     - com.sun.activation-javax.activation-1.2.0.jar
-    - javax.xml.bind-jaxb-api-2.3.1.jar
  * Java Servlet API -- javax.servlet-javax.servlet-api-3.1.0.jar
  * WebSocket Server API -- javax.websocket-javax.websocket-client-api-1.0.jar
- * Java Web Service REST API -- javax.ws.rs-javax.ws.rs-api-2.1.jar
  * HK2 - Dependency Injection Kernel
     - org.glassfish.hk2-hk2-api-2.6.1.jar
     - org.glassfish.hk2-hk2-locator-2.6.1.jar

distribution/shell/src/assemble/LICENSE.bin.txt

@@ -388,9 +388,9 @@ The Apache Software License, Version 2.0
     - log4j-slf4j2-impl-2.23.1.jar
     - log4j-web-2.23.1.jar
  * OpenTelemetry
-    - opentelemetry-api-1.41.0.jar
-    - opentelemetry-api-incubator-1.41.0-alpha.jar
-    - opentelemetry-context-1.41.0.jar
+    - opentelemetry-api-1.44.1.jar
+    - opentelemetry-api-incubator-1.44.1-alpha.jar
+    - opentelemetry-context-1.44.1.jar
 
  * BookKeeper
     - bookkeeper-common-allocator-4.17.1.jar
@@ -431,11 +431,8 @@ MIT License
 
 CDDL-1.1 -- ../licenses/LICENSE-CDDL-1.1.txt
  * Java Annotations API
-    - javax.annotation-api-1.3.2.jar
     - javax.activation-1.2.0.jar
-    - jaxb-api-2.3.1.jar
  * WebSocket Server API -- javax.websocket-client-api-1.0.jar
- * Java Web Service REST API -- javax.ws.rs-api-2.1.jar
  * HK2 - Dependency Injection Kernel
     - hk2-api-2.6.1.jar
     - hk2-locator-2.6.1.jar

jetcd-core-shaded/pom.xml

@@ -45,6 +45,10 @@
           <groupId>io.netty</groupId>
           <artifactId>*</artifactId>
         </exclusion>
+        <exclusion>
+          <groupId>javax.annotation</groupId>
+          <artifactId>javax.annotation-api</artifactId>
+        </exclusion>
       </exclusions>
     </dependency>
     <dependency>

pulsar-client/src/main/java/org/apache/pulsar/client/util/NameUtil.java → managed-ledger/src/main/java/org/apache/bookkeeper/mledger/CachedEntry.java

@@ -16,13 +16,11 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.client.util;
+package org.apache.bookkeeper.mledger;
 
-import java.util.UUID;
-import org.apache.commons.codec.digest.DigestUtils;
+import io.netty.util.ReferenceCounted;
 
-public class NameUtil {
-    public static String generateRandomName() {
-        return DigestUtils.sha1Hex(UUID.randomUUID().toString()).substring(0, 5);
-    }
+public interface CachedEntry extends Entry, ReferenceCounted {
+    boolean matchesKey(Position key);
+    boolean canEvict();
 }

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

@@ -32,8 +32,10 @@ public interface Entry {
     /**
      * @return the data
      */
+    @Deprecated
     byte[] getData();
 
+    @Deprecated
     byte[] getDataAndRelease();
 
     /**
@@ -66,4 +68,8 @@ public interface Entry {
      * of data reached to 0).
      */
     boolean release();
+
+    default EntryReadCountHandler getReadCountHandler() {
+        return null;
+    }
 }

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

@@ -0,0 +1,25 @@
+/*
+ * 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;
+
+public interface EntryReadCountHandler {
+    int getExpectedReadCount();
+    boolean incrementExpectedReadCount();
+    void markRead();
+}

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

@@ -19,10 +19,13 @@
 package org.apache.bookkeeper.mledger;
 
 import com.google.common.collect.Range;
+import java.util.Collection;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.SortedSet;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Predicate;
 import org.apache.bookkeeper.common.annotation.InterfaceAudience;
@@ -693,6 +696,7 @@ void asyncFindNewestMatching(FindPositionConstraint constraint, Predicate<Entry>
      * @throws InterruptedException
      * @throws ManagedLedgerException
      */
+    @Deprecated
     List<Entry> replayEntries(Set<? extends Position> positions)
             throws InterruptedException, ManagedLedgerException;
 
@@ -708,8 +712,10 @@ List<Entry> replayEntries(Set<? extends Position> positions)
      * @return skipped positions
      *              set of positions which are already deleted/acknowledged and skipped while replaying them
      */
-    Set<? extends Position> asyncReplayEntries(
-        Set<? extends Position> positions, ReadEntriesCallback callback, Object ctx);
+    default Set<? extends Position> asyncReplayEntries(final Set<? extends Position> positions,
+                                                      ReadEntriesCallback callback, Object ctx) {
+        return asyncReplayEntries(positions, callback, ctx, false);
+    }
 
     /**
      * Read the specified set of positions from ManagedLedger.
@@ -728,6 +734,28 @@ Set<? extends Position> asyncReplayEntries(
     Set<? extends Position> asyncReplayEntries(
             Set<? extends Position> positions, ReadEntriesCallback callback, Object ctx, boolean sortEntries);
 
+    /**
+     * Read the specified set of positions from ManagedLedger in ranges.
+     * This method is used to read entries in ranges to avoid reading all entries at once.
+     *
+     * @param positions
+     *            set of positions to read
+     * @param callback
+     *            callback object returning the result of each range
+     * @param ctx
+     *            opaque context
+     * @param invokeCallbacksInOrder
+     *            when true, the callback will be invoked in order of the positions, otherwise the callback will be
+     *            invoked in the order of the completion of the range read.
+     * @return skipped positions
+     *              set of positions which are already deleted/acknowledged and skipped while replaying them
+     */
+    default Set<? extends Position> asyncReplayEntriesInRanges(SortedSet<? extends Position> positions,
+                                                               ManagedCursorReplayReadEntriesCallback callback,
+                                                               Object ctx, boolean invokeCallbacksInOrder) {
+        throw new UnsupportedOperationException("Not implemented");
+    }
+
     /**
      * Close the cursor and releases the associated resources.
      *
@@ -893,6 +921,24 @@ default ManagedCursorAttributes getManagedCursorAttributes() {
 
     boolean isMessageDeleted(Position position);
 
+    /**
+     * Returns the deleted messages from the given positions.
+     * Implementation classes can override this method to provide a more efficient way to filter deleted messages.
+     *
+     * @param positions the positions to filter
+     * @return the set of deleted positions
+     */
+    default Set<Position> filterDeletedMessages(Collection<? extends Position> positions) {
+        Set<Position> deletedPositions = new HashSet<>();
+        // prefer for loop to avoid creating stream related instances
+        for (Position position : positions) {
+            if (isMessageDeleted(position)) {
+                deletedPositions.add(position);
+            }
+        }
+        return deletedPositions;
+    }
+
     ManagedCursor duplicateNonDurableCursor(String nonDurableCursorName) throws ManagedLedgerException;
 
     long[] getBatchPositionAckSet(Position position);

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

@@ -0,0 +1,28 @@
+/*
+ * 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;
+
+import java.util.List;
+
+public interface ManagedCursorReplayReadEntriesCallback {
+    void readEntriesComplete(ManagedCursorReplayReadRange range, boolean isLast, List<Entry> entries, Object ctx);
+
+    void readEntriesFailed(ManagedCursorReplayReadRange range, boolean isLast, ManagedLedgerException exception,
+                           Object ctx);
+}

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

@@ -0,0 +1,41 @@
+/*
+ * 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;
+
+public interface ManagedCursorReplayReadRange extends Comparable<ManagedCursorReplayReadRange> {
+    int rangeIndex();
+
+    int totalRanges();
+
+    Position startPosition();
+
+    Position lastPosition();
+
+    default int size() {
+        if (startPosition().getLedgerId() != lastPosition().getLedgerId()) {
+            throw new IllegalStateException("Cannot calculate size for range spanning multiple ledgers");
+        }
+        return (int) (lastPosition().getEntryId() - startPosition().getEntryId() + 1);
+    }
+
+    @Override
+    default int compareTo(ManagedCursorReplayReadRange o) {
+        return Integer.compare(rangeIndex(), o.rangeIndex());
+    }
+}

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

@@ -83,6 +83,9 @@ public class ManagedLedgerConfig {
     @Getter
     @Setter
     private boolean cacheEvictionByMarkDeletedPosition = false;
+    @Getter
+    @Setter
+    private boolean cacheEvictionByExpectedReadCount = true;
     private int minimumBacklogCursorsForCaching = 0;
     private int minimumBacklogEntriesForCaching = 1000;
     private int maxBacklogBetweenCursorsForCaching = 1000;

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

@@ -61,6 +61,18 @@ public class ManagedLedgerFactoryConfig {
      */
     private long managedLedgerMaxReadsInFlightSize = 0;
 
+    /**
+     * Maximum time to wait for acquiring permits for max reads in flight when managedLedgerMaxReadsInFlightSizeInMB is
+     * set (>0) and the limit is reached.
+     */
+    private long managedLedgerMaxReadsInFlightPermitsAcquireTimeoutMillis = 60000;
+
+    /**
+     * Maximum number of reads that can be queued for acquiring permits for max reads in flight when
+     * managedLedgerMaxReadsInFlightSizeInMB is set (>0) and the limit is reached.
+     */
+    private int managedLedgerMaxReadsInFlightPermitsAcquireQueueSize = 10000;
+
     /**
      * Whether trace managed ledger task execution time.
      */