Cherry pick/limit the memory

buildtools/pom.xml

@@ -269,18 +269,6 @@
           </execution>
         </executions>
       </plugin>
-      <plugin>
-        <artifactId>maven-gpg-plugin</artifactId>
-        <executions>
-          <execution>
-            <id>sign-artifacts</id>
-            <phase>verify</phase>
-            <goals>
-              <goal>sign</goal>
-            </goals>
-          </execution>
-        </executions>
-      </plugin>
     </plugins>
     <extensions>
       <extension>

conf/broker.conf

@@ -1064,7 +1064,15 @@ managedLedgerCursorRolloverTimeInSeconds=14400
 # crashes.
 managedLedgerMaxUnackedRangesToPersist=10000
 
-# Max number of "acknowledgment holes" that can be stored in Zookeeper. If number of unack message range is higher
+# Maximum amount of memory used hold data read from storage (or from the cache).
+# This mechanism prevents the broker to have too many concurrent
+# reads from storage and fall into Out of Memory errors in case
+# of multiple concurrent reads to multiple concurrent consumers.
+# Set 0 in order to disable the feature.
+#
+managedLedgerMaxReadsInFlightSizeInMB=0
+
+# Max number of "acknowledgment holes" that can be stored in MetadataStore. If number of unack message range is higher
 # than this limit then broker will persist unacked ranges into bookkeeper to avoid additional data overhead into
 # zookeeper.
 managedLedgerMaxUnackedRangesToPersistInZooKeeper=1000

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

@@ -57,6 +57,11 @@ public class ManagedLedgerFactoryConfig {
      */
     private boolean copyEntriesInCache = false;
 
+    /**
+     * Maximum number of (estimated) data in-flight reading from storage and the cache.
+     */
+    private long managedLedgerMaxReadsInFlightSize = 0;
+
     /**
      * Whether trace managed ledger task execution time.
      */

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

@@ -45,6 +45,8 @@ protected EntryImpl newObject(Handle<EntryImpl> handle) {
     private long entryId;
     ByteBuf data;
 
+    private Runnable onDeallocate;
+
     public static EntryImpl create(LedgerEntry ledgerEntry) {
         EntryImpl entry = RECYCLER.get();
         entry.timestamp = System.nanoTime();
@@ -103,6 +105,22 @@ private EntryImpl(Recycler.Handle<EntryImpl> recyclerHandle) {
         this.recyclerHandle = recyclerHandle;
     }
 
+    public void onDeallocate(Runnable r) {
+        if (this.onDeallocate == null) {
+            this.onDeallocate = r;
+        } else {
+            // this is not expected to happen
+            Runnable previous = this.onDeallocate;
+            this.onDeallocate = () -> {
+                try {
+                    previous.run();
+                } finally {
+                    r.run();
+                }
+            };
+        }
+    }
+
     public long getTimestamp() {
         return timestamp;
     }
@@ -160,6 +178,13 @@ public ReferenceCounted touch(Object hint) {
     @Override
     protected void deallocate() {
         // This method is called whenever the ref-count of the EntryImpl reaches 0, so that now we can recycle it
+        if (onDeallocate != null) {
+            try {
+                onDeallocate.run();
+            } finally {
+                onDeallocate = null;
+            }
+        }
         data.release();
         data = null;
         timestamp = -1;

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

@@ -0,0 +1,137 @@
+/**
+ * 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.cache;
+
+import com.google.common.annotations.VisibleForTesting;
+import io.prometheus.client.Gauge;
+import lombok.AllArgsConstructor;
+import lombok.ToString;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class InflightReadsLimiter {
+
+    private static final Gauge PULSAR_ML_READS_BUFFER_SIZE = Gauge
+            .build()
+            .name("pulsar_ml_reads_inflight_bytes")
+            .help("Estimated number of bytes retained by data read from storage or cache")
+            .register();
+
+    private static final Gauge PULSAR_ML_READS_AVAILABLE_BUFFER_SIZE = Gauge
+            .build()
+            .name("pulsar_ml_reads_available_inflight_bytes")
+            .help("Available space for inflight data read from storage or cache")
+            .register();
+
+    private final long maxReadsInFlightSize;
+    private long remainingBytes;
+
+    public InflightReadsLimiter(long maxReadsInFlightSize) {
+        if (maxReadsInFlightSize <= 0) {
+            // set it to -1 in order to show in the metrics that the metric is not available
+            PULSAR_ML_READS_BUFFER_SIZE.set(-1);
+            PULSAR_ML_READS_AVAILABLE_BUFFER_SIZE.set(-1);
+        }
+        this.maxReadsInFlightSize = maxReadsInFlightSize;
+        this.remainingBytes = maxReadsInFlightSize;
+    }
+
+    @VisibleForTesting
+    public synchronized long getRemainingBytes() {
+        return remainingBytes;
+    }
+
+    @AllArgsConstructor
+    @ToString
+    static class Handle {
+        final long acquiredPermits;
+        final boolean success;
+        final int trials;
+
+        final long creationTime;
+    }
+
+    private static final Handle DISABLED = new Handle(0, true, 0, -1);
+
+    Handle acquire(long permits, Handle current) {
+        if (maxReadsInFlightSize <= 0) {
+            // feature is disabled
+            return DISABLED;
+        }
+        synchronized (this) {
+            try {
+                if (current == null) {
+                    if (remainingBytes == 0) {
+                        return new Handle(0, false, 1, System.currentTimeMillis());
+                    }
+                    if (remainingBytes >= permits) {
+                        remainingBytes -= permits;
+                        return new Handle(permits, true, 1, System.currentTimeMillis());
+                    } else {
+                        long possible = remainingBytes;
+                        remainingBytes = 0;
+                        return new Handle(possible, false, 1, System.currentTimeMillis());
+                    }
+                } else {
+                    if (current.trials >= 4 && current.acquiredPermits > 0) {
+                        remainingBytes += current.acquiredPermits;
+                        return new Handle(0, false, 1, current.creationTime);
+                    }
+                    if (remainingBytes == 0) {
+                        return new Handle(current.acquiredPermits, false, current.trials + 1,
+                                current.creationTime);
+                    }
+                    long needed = permits - current.acquiredPermits;
+                    if (remainingBytes >= needed) {
+                        remainingBytes -= needed;
+                        return new Handle(permits, true, current.trials + 1, current.creationTime);
+                    } else {
+                        long possible = remainingBytes;
+                        remainingBytes = 0;
+                        return new Handle(current.acquiredPermits + possible, false,
+                                current.trials + 1, current.creationTime);
+                    }
+                }
+            } finally {
+                updateMetrics();
+            }
+        }
+    }
+
+    void release(Handle handle) {
+        if (handle == DISABLED) {
+            return;
+        }
+        synchronized (this) {
+            remainingBytes += handle.acquiredPermits;
+            updateMetrics();
+        }
+    }
+
+    private synchronized void updateMetrics() {
+        PULSAR_ML_READS_BUFFER_SIZE.set(maxReadsInFlightSize - remainingBytes);
+        PULSAR_ML_READS_AVAILABLE_BUFFER_SIZE.set(remainingBytes);
+    }
+
+    public boolean isDisabled() {
+        return maxReadsInFlightSize <= 0;
+    }
+
+
+}