[PIP 81]: Part-2 Add range set wrapper to support record dirty ledgers

conf/broker.conf

@@ -1480,3 +1480,7 @@ subscriptionKeySharedEnable=true
 # zookeeper.
 # Deprecated: use managedLedgerMaxUnackedRangesToPersistInMetadataStore
 managedLedgerMaxUnackedRangesToPersistInZooKeeper=-1
+
+# If enabled, the maximum "acknowledgment holes" will not be limited and "acknowledgment holes" are stored in
+# multiple entries.
+persistentUnackedRangesWithMultipleEntriesEnabled=false

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

@@ -44,6 +44,7 @@ public class ManagedLedgerConfig {
     private boolean createIfMissing = true;
     private int maxUnackedRangesToPersist = 10000;
     private int maxBatchDeletedIndexToPersist = 10000;
+    private boolean persistentUnackedRangesWithMultipleEntriesEnabled = false;
     private boolean deletionAtBatchIndexLevelEnabled = true;
     private int maxUnackedRangesToPersistInMetadataStore = 1000;
     private int maxEntriesPerLedger = 50000;
@@ -470,6 +471,14 @@ public int getMaxBatchDeletedIndexToPersist() {
         return maxBatchDeletedIndexToPersist;
     }
 
+    public boolean isPersistentUnackedRangesWithMultipleEntriesEnabled() {
+        return persistentUnackedRangesWithMultipleEntriesEnabled;
+    }
+
+    public void setPersistentUnackedRangesWithMultipleEntriesEnabled(boolean multipleEntriesEnabled) {
+        this.persistentUnackedRangesWithMultipleEntriesEnabled = multipleEntriesEnabled;
+    }
+
     /**
      * @param maxUnackedRangesToPersist
      *            max unacked message ranges that will be persisted and receverd.

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

@@ -93,7 +93,6 @@
 import org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.pulsar.common.util.collections.BitSetRecyclable;
-import org.apache.pulsar.common.util.collections.ConcurrentOpenLongPairRangeSet;
 import org.apache.pulsar.common.util.collections.LongPairRangeSet;
 import org.apache.pulsar.common.util.collections.LongPairRangeSet.LongPairConsumer;
 import org.apache.pulsar.metadata.api.Stat;
@@ -180,7 +179,7 @@ public class ManagedCursorImpl implements ManagedCursor {
         position.ackSet = null;
         return position;
     };
-    private final LongPairRangeSet<PositionImpl> individualDeletedMessages;
+    private final RangeSetWrapper<PositionImpl> individualDeletedMessages;
 
     // Maintain the deletion status for batch messages
     // (ledgerId, entryId) -> deletion indexes
@@ -284,9 +283,7 @@ public interface VoidCallback {
         this.config = config;
         this.ledger = ledger;
         this.name = cursorName;
-        this.individualDeletedMessages = config.isUnackedRangesOpenCacheSetEnabled()
-                ? new ConcurrentOpenLongPairRangeSet<>(4096, positionRangeConverter)
-                : new LongPairRangeSet.DefaultRangeSet<>(positionRangeConverter);
+        this.individualDeletedMessages = new RangeSetWrapper<>(positionRangeConverter, this);
         if (config.isDeletionAtBatchIndexLevelEnabled()) {
             this.batchDeletedIndexes = new ConcurrentSkipListMap<>();
         } else {
@@ -2649,6 +2646,7 @@ private List<MLDataFormats.MessageRange> buildIndividualDeletedMessageRanges() {
                 return rangeList.size() <= config.getMaxUnackedRangesToPersist();
             });
             this.individualDeletedMessagesSerializedSize = acksSerializedSize.get();
+            individualDeletedMessages.resetDirtyKeys();
             return rangeList;
         } finally {
             lock.readLock().unlock();
@@ -3196,4 +3194,8 @@ public void setState(State state) {
     }
 
     private static final Logger log = LoggerFactory.getLogger(ManagedCursorImpl.class);
+
+    public ManagedLedgerConfig getConfig() {
+        return config;
+    }
 }

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

@@ -0,0 +1,165 @@
+/**
+ * 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 com.google.common.base.Preconditions.checkNotNull;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Range;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenLongPairRangeSet;
+import org.apache.pulsar.common.util.collections.LongPairRangeSet;
+
+/**
+ * Wraps other Range classes, and adds LRU, marking dirty data and other features on this basis.
+ * This range set is not thread safety.
+ *
+ * @param <T>
+ */
+public class RangeSetWrapper<T extends Comparable<T>> implements LongPairRangeSet<T> {
+
+    private final LongPairRangeSet<T> rangeSet;
+    private final LongPairConsumer<T> rangeConverter;
+    private final ManagedLedgerConfig config;
+    private final boolean enableMultiEntry;
+
+    /**
+     * Record which Ledger is dirty.
+     */
+    private final DefaultRangeSet<Long> dirtyLedgers = new LongPairRangeSet.DefaultRangeSet<>(
+            (LongPairConsumer<Long>) (key, value) -> key);
+
+    public RangeSetWrapper(LongPairConsumer<T> rangeConverter, ManagedCursorImpl managedCursor) {
+        checkNotNull(managedCursor);
+        this.config = managedCursor.getConfig();
+        this.rangeConverter = rangeConverter;
+        this.rangeSet = config.isUnackedRangesOpenCacheSetEnabled()
+                ? new ConcurrentOpenLongPairRangeSet<>(4096, rangeConverter)
+                : new LongPairRangeSet.DefaultRangeSet<>(rangeConverter);
+        this.enableMultiEntry = config.isPersistentUnackedRangesWithMultipleEntriesEnabled();
+    }
+
+    @Override
+    public void addOpenClosed(long lowerKey, long lowerValue, long upperKey, long upperValue) {
+        if (enableMultiEntry) {
+            dirtyLedgers.addOpenClosed(lowerKey, 0, upperKey, 0);
+        }
+        rangeSet.addOpenClosed(lowerKey, lowerValue, upperKey, upperValue);
+    }
+
+    @Override
+    public boolean contains(long key, long value) {
+        return rangeSet.contains(key, value);
+    }
+
+    @Override
+    public Range<T> rangeContaining(long key, long value) {
+        return rangeSet.rangeContaining(key, value);
+    }
+
+    @Override
+    public void removeAtMost(long key, long value) {
+        if (enableMultiEntry) {
+            dirtyLedgers.removeAtMost(key, 0);
+        }
+        rangeSet.removeAtMost(key, value);
+    }
+
+    @Override
+    public boolean isEmpty() {
+        return rangeSet.isEmpty();
+    }
+
+    @Override
+    public void clear() {
+        rangeSet.clear();
+        dirtyLedgers.clear();
+    }
+
+    @Override
+    public Range<T> span() {
+        return rangeSet.span();
+    }
+
+    @Override
+    public Collection<Range<T>> asRanges() {
+        Collection<Range<T>> collection = rangeSet.asRanges();
+        if (collection instanceof List) {
+            return collection;
+        }
+        return new ArrayList<>(collection);
+    }
+
+    @Override
+    public void forEach(RangeProcessor<T> action) {
+        rangeSet.forEach(action);
+    }
+
+    @Override
+    public void forEach(RangeProcessor<T> action, LongPairConsumer<? extends T> consumer) {
+        rangeSet.forEach(action, consumer);
+    }
+
+    @Override
+    public int size() {
+        return rangeSet.size();
+    }
+
+    @Override
+    public Range<T> firstRange() {
+        return rangeSet.firstRange();
+    }
+
+    @Override
+    public Range<T> lastRange() {
+        return rangeSet.lastRange();
+    }
+
+    @VisibleForTesting
+    void add(Range<LongPair> range) {
+        if (!(rangeSet instanceof ConcurrentOpenLongPairRangeSet)) {
+            throw new UnsupportedOperationException("Only ConcurrentOpenLongPairRangeSet support this method");
+        }
+        ((ConcurrentOpenLongPairRangeSet<T>) rangeSet).add(range);
+    }
+
+    @VisibleForTesting
+    void remove(Range<T> range) {
+        if (rangeSet instanceof ConcurrentOpenLongPairRangeSet) {
+            ((ConcurrentOpenLongPairRangeSet<T>) rangeSet).remove((Range<LongPair>) range);
+        } else {
+            ((DefaultRangeSet<T>) rangeSet).remove(range);
+        }
+    }
+
+    public void resetDirtyKeys() {
+        dirtyLedgers.clear();
+    }
+
+    public boolean isDirtyLedgers(long ledgerId) {
+        return dirtyLedgers.contains(ledgerId);
+    }
+
+    @Override
+    public String toString() {
+        return rangeSet.toString();
+    }
+}