You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by rd...@apache.org on 2019/05/21 18:09:48 UTC
[pulsar] branch master updated: [pulsar-broker] Cache
unack-messageId into openRangeSet (#3819)
This is an automated email from the ASF dual-hosted git repository.
rdhabalia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 5556f1b [pulsar-broker] Cache unack-messageId into openRangeSet (#3819)
5556f1b is described below
commit 5556f1b7e01040bd78c7e7736383812eb12217ad
Author: Rajan Dhabalia <rd...@apache.org>
AuthorDate: Tue May 21 11:09:43 2019 -0700
[pulsar-broker] Cache unack-messageId into openRangeSet (#3819)
* [pulsar-common] add open Concurrent LongPair RangeSet
* add open-range set methods
* [pulsar-broker] Cache unack-messageId into openRangeSet
* add forEach
* use forEach of the range
* add forEach with consumer
* add Recyclable Position converter
* Removed unused imports after merge
* address comments
---
conf/broker.conf | 3 +
conf/standalone.conf | 3 +
.../bookkeeper/mledger/ManagedLedgerConfig.java | 15 +-
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 200 ++++++++++++-------
.../bookkeeper/mledger/impl/PositionImpl.java | 8 +-
.../mledger/impl/PositionImplRecyclable.java | 50 +++++
.../mledger/impl/ManagedCursorConcurrencyTest.java | 14 +-
.../bookkeeper/mledger/impl/ManagedCursorTest.java | 132 ++++++++----
.../mledger/impl/ManagedLedgerFactoryTest.java | 10 +-
.../apache/pulsar/broker/ServiceConfiguration.java | 6 +-
.../pulsar/broker/service/BrokerService.java | 2 +
.../pulsar/client/impl/MessageRedeliveryTest.java | 221 +++++++++++++++++++++
12 files changed, 544 insertions(+), 120 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index 8596e14..f6c8221 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -683,6 +683,9 @@ managedLedgerOffloadDriver=
# Maximum number of thread pool threads for ledger offloading
managedLedgerOffloadMaxThreads=2
+# Use Open Range-Set to cache unacked messages
+managedLedgerUnackedRangesOpenCacheSetEnabled=true
+
# For Amazon S3 ledger offload, AWS region
s3ManagedLedgerOffloadRegion=
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 7e8f692..46a6d77 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -371,6 +371,9 @@ managedLedgerReadEntryTimeoutSeconds=120
# Add entry timeout when broker tries to publish message to bookkeeper (0 to disable it).
managedLedgerAddEntryTimeoutSeconds=120
+# Use Open Range-Set to cache unacked messages
+managedLedgerUnackedRangesOpenCacheSetEnabled=true
+
### --- Load balancer --- ###
loadManagerClassName=org.apache.pulsar.broker.loadbalance.NoopLoadManager
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
index 4a57055..0e77aff 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
@@ -28,7 +28,6 @@ import java.util.concurrent.TimeUnit;
import org.apache.bookkeeper.client.api.DigestType;
import org.apache.bookkeeper.mledger.impl.NullLedgerOffloader;
-
/**
* Configuration class for a ManagedLedger.
*/
@@ -61,6 +60,7 @@ public class ManagedLedgerConfig {
private long addEntryTimeoutSeconds = 120;
private DigestType digestType = DigestType.CRC32C;
private byte[] password = "".getBytes(Charsets.UTF_8);
+ private boolean unackedRangesOpenCacheSetEnabled = true;
private LedgerOffloader ledgerOffloader = NullLedgerOffloader.INSTANCE;
private Clock clock = Clock.systemUTC();
@@ -236,6 +236,19 @@ public class ManagedLedgerConfig {
}
/**
+ * should use {@link ConcurrentOpenLongPairRangeSet} to store unacked ranges.
+ * @return
+ */
+ public boolean isUnackedRangesOpenCacheSetEnabled() {
+ return unackedRangesOpenCacheSetEnabled;
+ }
+
+ public ManagedLedgerConfig setUnackedRangesOpenCacheSetEnabled(boolean unackedRangesOpenCacheSetEnabled) {
+ this.unackedRangesOpenCacheSetEnabled = unackedRangesOpenCacheSetEnabled;
+ return this;
+ }
+
+ /**
* @return the metadataEnsemblesize
*/
public int getMetadataEnsemblesize() {
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index 54521e3..cdad87d 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -26,18 +26,19 @@ import static org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.DEFAULT_LEDGE
import static org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.createManagedLedgerException;
import static org.apache.bookkeeper.mledger.util.SafeRun.safeRun;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import com.google.common.base.Predicate;
import com.google.common.collect.Collections2;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Range;
-import com.google.common.collect.RangeSet;
import com.google.common.collect.Sets;
-import com.google.common.collect.TreeRangeSet;
import com.google.common.util.concurrent.RateLimiter;
import com.google.protobuf.InvalidProtocolBufferException;
+import io.netty.util.concurrent.FastThreadLocal;
+
import java.time.Clock;
import java.util.ArrayDeque;
import java.util.Collections;
@@ -48,11 +49,11 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
-import java.util.stream.Collectors;
import org.apache.bookkeeper.client.AsyncCallback.CloseCallback;
import org.apache.bookkeeper.client.AsyncCallback.DeleteCallback;
@@ -81,8 +82,12 @@ import org.apache.bookkeeper.mledger.impl.MetaStore.Stat;
import org.apache.bookkeeper.mledger.proto.MLDataFormats;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.common.util.collections.LongPairRangeSet;
+import org.apache.pulsar.common.util.collections.LongPairRangeSet.LongPairConsumer;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenLongPairRangeSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -125,12 +130,26 @@ public class ManagedCursorImpl implements ManagedCursor {
// Stat of the cursor z-node
private volatile Stat cursorLedgerStat;
- private final RangeSet<PositionImpl> individualDeletedMessages = TreeRangeSet.create();
+ private static final LongPairConsumer<PositionImpl> positionRangeConverter = (key, value) -> new PositionImpl(key,
+ value);
+ private static final LongPairConsumer<PositionImplRecyclable> recyclePositionRangeConverter = (key, value) -> {
+ PositionImplRecyclable position = PositionImplRecyclable.create();
+ position.ledgerId = key;
+ position.entryId = value;
+ return position;
+ };
+ private final LongPairRangeSet<PositionImpl> individualDeletedMessages;
private final ReadWriteLock lock = new ReentrantReadWriteLock();
private RateLimiter markDeleteLimiter;
private boolean alwaysInactive = false;
+
+ /** used temporary variables to {@link #getNumIndividualDeletedEntriesToSkip(long)} **/
+ private static final FastThreadLocal<Long> tempTotalEntriesToSkip = new FastThreadLocal<>();
+ private static final FastThreadLocal<Long> tempDeletedMessages = new FastThreadLocal<>();
+ private static final FastThreadLocal<PositionImpl> tempStartPosition = new FastThreadLocal<>();
+ private static final FastThreadLocal<PositionImpl> tempEndPosition = new FastThreadLocal<>();
class MarkDeleteEntry {
final PositionImpl newPosition;
@@ -188,6 +207,9 @@ public class ManagedCursorImpl implements ManagedCursor {
this.config = config;
this.ledger = ledger;
this.name = cursorName;
+ this.individualDeletedMessages = config.isUnackedRangesOpenCacheSetEnabled()
+ ? new ConcurrentOpenLongPairRangeSet<PositionImpl>(4096, positionRangeConverter)
+ : new LongPairRangeSet.DefaultRangeSet<>(positionRangeConverter);
this.digestType = BookKeeper.DigestType.fromApiDigestType(config.getDigestType());
STATE_UPDATER.set(this, State.Uninitialized);
PENDING_MARK_DELETED_SUBMITTED_COUNT_UPDATER.set(this, 0);
@@ -346,8 +368,9 @@ public class ManagedCursorImpl implements ManagedCursor {
try {
individualDeletedMessages.clear();
individualDeletedMessagesList.forEach(messageRange -> individualDeletedMessages
- .add(Range.openClosed(new PositionImpl(messageRange.getLowerEndpoint()),
- new PositionImpl(messageRange.getUpperEndpoint()))));
+ .addOpenClosed(messageRange.getLowerEndpoint().getLedgerId(),
+ messageRange.getLowerEndpoint().getEntryId(), messageRange.getUpperEndpoint().getLedgerId(),
+ messageRange.getUpperEndpoint().getEntryId()));
} finally {
lock.writeLock().unlock();
}
@@ -692,7 +715,7 @@ public class ManagedCursorImpl implements ManagedCursor {
@Override
public int getTotalNonContiguousDeletedMessagesRange() {
- return individualDeletedMessages.asRanges().size();
+ return individualDeletedMessages.size();
}
@Override
@@ -1014,7 +1037,8 @@ public class ManagedCursorImpl implements ManagedCursor {
lock.readLock().lock();
try {
positions.stream()
- .filter(position -> individualDeletedMessages.contains((PositionImpl) position)
+ .filter(position -> individualDeletedMessages.contains(((PositionImpl) position).getLedgerId(),
+ ((PositionImpl) position).getEntryId())
|| ((PositionImpl) position).compareTo(markDeletePosition) < 0)
.forEach(alreadyAcknowledgedPositions::add);
} finally {
@@ -1069,30 +1093,39 @@ public class ManagedCursorImpl implements ManagedCursor {
log.debug("[{}] getNumberOfEntries. {} allEntries: {}", ledger.getName(), range, allEntries);
}
- long deletedEntries = 0;
+ AtomicLong deletedEntries = new AtomicLong(0);
lock.readLock().lock();
try {
- for (Range<PositionImpl> r : individualDeletedMessages.asRanges()) {
- if (r.isConnected(range)) {
- Range<PositionImpl> commonEntries = r.intersection(range);
- long commonCount = ledger.getNumberOfEntries(commonEntries);
- if (log.isDebugEnabled()) {
- log.debug("[{}] [{}] Discounting {} entries for already deleted range {}", ledger.getName(),
- name, commonCount, commonEntries);
+ individualDeletedMessages.forEach((r) -> {
+ try {
+ if (r.isConnected(range)) {
+ Range<PositionImpl> commonEntries = r.intersection(range);
+ long commonCount = ledger.getNumberOfEntries(commonEntries);
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] [{}] Discounting {} entries for already deleted range {}", ledger.getName(),
+ name, commonCount, commonEntries);
+ }
+ deletedEntries.addAndGet(commonCount);
+ }
+ return true;
+ } finally {
+ if (r.lowerEndpoint() instanceof PositionImplRecyclable) {
+ ((PositionImplRecyclable) r.lowerEndpoint()).recycle();
+ ((PositionImplRecyclable) r.upperEndpoint()).recycle();
}
- deletedEntries += commonCount;
}
- }
+ }, recyclePositionRangeConverter);
} finally {
lock.readLock().unlock();
}
if (log.isDebugEnabled()) {
- log.debug("[{}] Found {} entries - deleted: {}",
- ledger.getName(), allEntries - deletedEntries, deletedEntries);
+ log.debug("[{}] Found {} entries - deleted: {}", ledger.getName(), allEntries - deletedEntries.get(),
+ deletedEntries);
}
- return allEntries - deletedEntries;
+ return allEntries - deletedEntries.get();
+
}
@Override
@@ -1252,34 +1285,43 @@ public class ManagedCursorImpl implements ManagedCursor {
}
long getNumIndividualDeletedEntriesToSkip(long numEntries) {
- long totalEntriesToSkip = 0;
- long deletedMessages = 0;
+ tempTotalEntriesToSkip.set(0L);
+ tempDeletedMessages.set(0L);
lock.readLock().lock();
try {
- PositionImpl startPosition = markDeletePosition;
- PositionImpl endPosition = null;
- for (Range<PositionImpl> r : individualDeletedMessages.asRanges()) {
- endPosition = r.lowerEndpoint();
- if (startPosition.compareTo(endPosition) <= 0) {
- Range<PositionImpl> range = Range.openClosed(startPosition, endPosition);
- long entries = ledger.getNumberOfEntries(range);
- if (totalEntriesToSkip + entries >= numEntries) {
- break;
+ tempStartPosition.set(markDeletePosition);
+ tempEndPosition.set(null);
+ individualDeletedMessages.forEach((r) -> {
+ try {
+ tempEndPosition.set(r.lowerEndpoint());
+ if (tempStartPosition.get().compareTo(tempEndPosition.get()) <= 0) {
+ Range<PositionImpl> range = Range.openClosed(tempStartPosition.get(), tempEndPosition.get());
+ long entries = ledger.getNumberOfEntries(range);
+ if (tempTotalEntriesToSkip.get() + entries >= numEntries) {
+ // do not process further
+ return false;
+ }
+ tempTotalEntriesToSkip.set(tempTotalEntriesToSkip.get() + entries);
+ tempDeletedMessages.set(tempDeletedMessages.get() + ledger.getNumberOfEntries(r));
+ tempStartPosition.set(r.upperEndpoint());
+ } else {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] deletePosition {} moved ahead without clearing deleteMsgs {} for cursor {}",
+ ledger.getName(), markDeletePosition, r.lowerEndpoint(), name);
+ }
}
- totalEntriesToSkip += entries;
- deletedMessages += ledger.getNumberOfEntries(r);
- startPosition = r.upperEndpoint();
- } else {
- if (log.isDebugEnabled()) {
- log.debug("[{}] deletePosition {} moved ahead without clearing deleteMsgs {} for cursor {}",
- ledger.getName(), markDeletePosition, r.lowerEndpoint(), name);
+ return true;
+ } finally {
+ if (r.lowerEndpoint() instanceof PositionImplRecyclable) {
+ ((PositionImplRecyclable) r.lowerEndpoint()).recycle();
+ ((PositionImplRecyclable) r.upperEndpoint()).recycle();
}
}
- }
+ }, recyclePositionRangeConverter);
} finally {
lock.readLock().unlock();
}
- return deletedMessages;
+ return tempDeletedMessages.get();
}
boolean hasMoreEntries(PositionImpl position) {
@@ -1317,17 +1359,29 @@ public class ManagedCursorImpl implements ManagedCursor {
if (newMarkDeletePosition.getLedgerId() == oldMarkDeletePosition.getLedgerId()
&& newMarkDeletePosition.getEntryId() == oldMarkDeletePosition.getEntryId() + 1) {
// Mark-deleting the position next to current one
- skippedEntries = individualDeletedMessages.contains(newMarkDeletePosition) ? 0 : 1;
+ skippedEntries = individualDeletedMessages.contains(newMarkDeletePosition.getLedgerId(),
+ newMarkDeletePosition.getEntryId()) ? 0 : 1;
} else {
skippedEntries = getNumberOfEntries(Range.openClosed(oldMarkDeletePosition, newMarkDeletePosition));
}
+
PositionImpl positionAfterNewMarkDelete = ledger.getNextValidPosition(newMarkDeletePosition);
- if (individualDeletedMessages.contains(positionAfterNewMarkDelete)) {
- Range<PositionImpl> rangeToBeMarkDeleted = individualDeletedMessages
- .rangeContaining(positionAfterNewMarkDelete);
- newMarkDeletePosition = rangeToBeMarkDeleted.upperEndpoint();
+ // sometime ranges are connected but belongs to different ledgers so, they are placed sequentially
+ // eg: (2:10..3:15] can be returned as (2:10..2:15],[3:0..3:15]. So, try to iterate over connected range and
+ // found the last non-connected range which gives new markDeletePosition
+ while (positionAfterNewMarkDelete.compareTo(ledger.lastConfirmedEntry) <= 0) {
+ if (individualDeletedMessages.contains(positionAfterNewMarkDelete.getLedgerId(),
+ positionAfterNewMarkDelete.getEntryId())) {
+ Range<PositionImpl> rangeToBeMarkDeleted = individualDeletedMessages.rangeContaining(
+ positionAfterNewMarkDelete.getLedgerId(), positionAfterNewMarkDelete.getEntryId());
+ newMarkDeletePosition = rangeToBeMarkDeleted.upperEndpoint();
+ positionAfterNewMarkDelete = ledger.getNextValidPosition(newMarkDeletePosition);
+ // check if next valid position is also deleted and part of the deleted-range
+ continue;
+ }
+ break;
}
-
+
if (log.isDebugEnabled()) {
log.debug("[{}] Moved ack position from: {} to: {} -- skipped: {}", ledger.getName(),
oldMarkDeletePosition, newMarkDeletePosition, skippedEntries);
@@ -1337,7 +1391,7 @@ public class ManagedCursorImpl implements ManagedCursor {
// markDelete-position and clear out deletedMsgSet
markDeletePosition = PositionImpl.get(newMarkDeletePosition);
- individualDeletedMessages.remove(Range.atMost(markDeletePosition));
+ individualDeletedMessages.removeAtMost(markDeletePosition.getLedgerId(), markDeletePosition.getEntryId());
if (readPosition.compareTo(newMarkDeletePosition) <= 0) {
// If the position that is mark-deleted is past the read position, it
@@ -1476,7 +1530,8 @@ public class ManagedCursorImpl implements ManagedCursor {
// point.
lock.writeLock().lock();
try {
- individualDeletedMessages.remove(Range.atMost(mdEntry.newPosition));
+ individualDeletedMessages.removeAtMost(mdEntry.newPosition.getLedgerId(),
+ mdEntry.newPosition.getEntryId());
} finally {
lock.writeLock().unlock();
}
@@ -1610,7 +1665,8 @@ public class ManagedCursorImpl implements ManagedCursor {
return;
}
- if (individualDeletedMessages.contains(position) || position.compareTo(markDeletePosition) <= 0) {
+ if (individualDeletedMessages.contains(position.getLedgerId(), position.getEntryId())
+ || position.compareTo(markDeletePosition) <= 0) {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Position was already deleted {}", ledger.getName(), name, position);
}
@@ -1620,7 +1676,8 @@ public class ManagedCursorImpl implements ManagedCursor {
// Add a range (prev, pos] to the set. Adding the previous entry as an open limit to the range will make
// the RangeSet recognize the "continuity" between adjacent Positions
PositionImpl previousPosition = ledger.getPreviousPosition(position);
- individualDeletedMessages.add(Range.openClosed(previousPosition, position));
+ individualDeletedMessages.addOpenClosed(previousPosition.getLedgerId(), previousPosition.getEntryId(),
+ position.getLedgerId(), position.getEntryId());
++messagesConsumedCounter;
if (log.isDebugEnabled()) {
@@ -1637,7 +1694,7 @@ public class ManagedCursorImpl implements ManagedCursor {
// If the lower bound of the range set is the current mark delete position, then we can trigger a new
// mark-delete to the upper bound of the first range segment
- Range<PositionImpl> range = individualDeletedMessages.asRanges().iterator().next();
+ Range<PositionImpl> range = individualDeletedMessages.firstRange();
// If the lowerBound is ahead of MarkDelete, verify if there are any entries in-between
if (range.lowerEndpoint().compareTo(markDeletePosition) <= 0 || ledger
@@ -1726,7 +1783,9 @@ public class ManagedCursorImpl implements ManagedCursor {
} else {
// Remove from the entry list all the entries that were already marked for deletion
return Lists.newArrayList(Collections2.filter(entries, entry -> {
- boolean includeEntry = !individualDeletedMessages.contains((PositionImpl) entry.getPosition());
+ boolean includeEntry = !individualDeletedMessages.contains(
+ ((PositionImpl) entry.getPosition()).getLedgerId(),
+ ((PositionImpl) entry.getPosition()).getEntryId());
if (!includeEntry) {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Filtering entry at {} - already deleted", ledger.getName(), name,
@@ -1899,7 +1958,7 @@ public class ManagedCursorImpl implements ManagedCursor {
private boolean shouldPersistUnackRangesToLedger() {
return cursorLedger != null && config.getMaxUnackedRangesToPersist() > 0
- && individualDeletedMessages.asRanges().size() > config.getMaxUnackedRangesToPersistInZk();
+ && individualDeletedMessages.size() > config.getMaxUnackedRangesToPersistInZk();
}
private void persistPositionMetaStore(long cursorsLedgerId, PositionImpl position, Map<String, Long> properties,
@@ -2124,18 +2183,20 @@ public class ManagedCursorImpl implements ManagedCursor {
MLDataFormats.NestedPositionInfo.Builder nestedPositionBuilder = MLDataFormats.NestedPositionInfo
.newBuilder();
MLDataFormats.MessageRange.Builder messageRangeBuilder = MLDataFormats.MessageRange.newBuilder();
- return individualDeletedMessages.asRanges().stream().limit(config.getMaxUnackedRangesToPersist())
- .map(positionRange -> {
- PositionImpl p = positionRange.lowerEndpoint();
- nestedPositionBuilder.setLedgerId(p.getLedgerId());
- nestedPositionBuilder.setEntryId(p.getEntryId());
- messageRangeBuilder.setLowerEndpoint(nestedPositionBuilder.build());
- p = positionRange.upperEndpoint();
- nestedPositionBuilder.setLedgerId(p.getLedgerId());
- nestedPositionBuilder.setEntryId(p.getEntryId());
- messageRangeBuilder.setUpperEndpoint(nestedPositionBuilder.build());
- return messageRangeBuilder.build();
- }).collect(Collectors.toList());
+ List<MessageRange> rangeList = Lists.newArrayList();
+ individualDeletedMessages.forEach((positionRange) -> {
+ PositionImpl p = positionRange.lowerEndpoint();
+ nestedPositionBuilder.setLedgerId(p.getLedgerId());
+ nestedPositionBuilder.setEntryId(p.getEntryId());
+ messageRangeBuilder.setLowerEndpoint(nestedPositionBuilder.build());
+ p = positionRange.upperEndpoint();
+ nestedPositionBuilder.setLedgerId(p.getLedgerId());
+ nestedPositionBuilder.setEntryId(p.getEntryId());
+ messageRangeBuilder.setUpperEndpoint(nestedPositionBuilder.build());
+ rangeList.add(messageRangeBuilder.build());
+ return rangeList.size() <= config.getMaxUnackedRangesToPersist();
+ });
+ return rangeList;
} finally {
lock.readLock().unlock();
}
@@ -2440,6 +2501,11 @@ public class ManagedCursorImpl implements ManagedCursor {
}
}
+ @VisibleForTesting
+ public LongPairRangeSet<PositionImpl> getIndividuallyDeletedMessagesSet() {
+ return individualDeletedMessages;
+ }
+
/**
* Checks given position is part of deleted-range and returns next position of upper-end as all the messages are
* deleted up to that point.
@@ -2448,7 +2514,7 @@ public class ManagedCursorImpl implements ManagedCursor {
* @return next available position
*/
public PositionImpl getNextAvailablePosition(PositionImpl position) {
- Range<PositionImpl> range = individualDeletedMessages.rangeContaining(position);
+ Range<PositionImpl> range = individualDeletedMessages.rangeContaining(position.getLedgerId(), position.getEntryId());
if (range != null) {
PositionImpl nextPosition = range.upperEndpoint().getNext();
return (nextPosition != null && nextPosition.compareTo(position) > 0) ? nextPosition : position.getNext();
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionImpl.java
index a0607a8..f090b52 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionImpl.java
@@ -28,11 +28,11 @@ import org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo;
public class PositionImpl implements Position, Comparable<PositionImpl> {
- private final long ledgerId;
- private final long entryId;
+ protected long ledgerId;
+ protected long entryId;
- public static PositionImpl earliest = new PositionImpl(-1, -1);
- public static PositionImpl latest = new PositionImpl(Long.MAX_VALUE, Long.MAX_VALUE);
+ public static final PositionImpl earliest = new PositionImpl(-1, -1);
+ public static final PositionImpl latest = new PositionImpl(Long.MAX_VALUE, Long.MAX_VALUE);
public PositionImpl(PositionInfo pi) {
this.ledgerId = pi.getLedgerId();
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionImplRecyclable.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionImplRecyclable.java
new file mode 100644
index 0000000..6011996
--- /dev/null
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionImplRecyclable.java
@@ -0,0 +1,50 @@
+/**
+ * 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 org.apache.bookkeeper.mledger.Position;
+
+import io.netty.util.Recycler;
+import io.netty.util.Recycler.Handle;
+
+public class PositionImplRecyclable extends PositionImpl implements Position {
+
+ private final Handle<PositionImplRecyclable> recyclerHandle;
+
+ private static final Recycler<PositionImplRecyclable> RECYCLER = new Recycler<PositionImplRecyclable>() {
+ protected PositionImplRecyclable newObject(Recycler.Handle<PositionImplRecyclable> recyclerHandle) {
+ return new PositionImplRecyclable(recyclerHandle);
+ }
+ };
+
+ private PositionImplRecyclable(Handle<PositionImplRecyclable> recyclerHandle) {
+ super(PositionImpl.earliest);
+ this.recyclerHandle = recyclerHandle;
+ }
+
+ public static PositionImplRecyclable create() {
+ PositionImplRecyclable position = RECYCLER.get();
+ return position;
+ }
+
+ public void recycle() {
+ recyclerHandle.recycle(this);
+ }
+
+}
\ No newline at end of file
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorConcurrencyTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorConcurrencyTest.java
index 62f4045..852abcd 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorConcurrencyTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorConcurrencyTest.java
@@ -42,11 +42,17 @@ import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
public class ManagedCursorConcurrencyTest extends MockedBookKeeperTestCase {
private static final Logger log = LoggerFactory.getLogger(ManagedCursorConcurrencyTest.class);
+
+ @DataProvider(name = "useOpenRangeSet")
+ public static Object[][] useOpenRangeSet() {
+ return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } };
+ }
private final AsyncCallbacks.DeleteCallback deleteCallback = new AsyncCallbacks.DeleteCallback() {
@Override
@@ -60,9 +66,11 @@ public class ManagedCursorConcurrencyTest extends MockedBookKeeperTestCase {
}
};
- @Test
- public void testMarkDeleteAndRead() throws Exception {
- ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(2));
+ @Test(dataProvider = "useOpenRangeSet")
+ public void testMarkDeleteAndRead(boolean useOpenRangeSet) throws Exception {
+ ManagedLedgerConfig config = new ManagedLedgerConfig().setMaxEntriesPerLedger(2)
+ .setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet);
+ ManagedLedger ledger = factory.open("my_test_ledger", config);
final ManagedCursor cursor = ledger.openCursor("c1");
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
index ef9bfb0..4592e32 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
@@ -73,16 +73,22 @@ import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo;
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
import org.apache.zookeeper.KeeperException.Code;
-import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
public class ManagedCursorTest extends MockedBookKeeperTestCase {
private static final Charset Encoding = Charsets.UTF_8;
+
+ @DataProvider(name = "useOpenRangeSet")
+ public static Object[][] useOpenRangeSet() {
+ return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } };
+ }
+
@Test(timeOut = 20000)
void readFromEmptyLedger() throws Exception {
@@ -1315,9 +1321,10 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
assertEquals(cursor.getNumberOfEntriesInBacklog(), 3);
}
- @Test(timeOut = 20000)
- void testMarkDeleteTwice() throws Exception {
- ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(2));
+ @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
+ void testMarkDeleteTwice(boolean useOpenRangeSet) throws Exception {
+ ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig()
+ .setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet).setMaxEntriesPerLedger(2));
ManagedCursor cursor = ledger.openCursor("c1");
Position p1 = ledger.addEntry("entry1".getBytes());
@@ -1327,9 +1334,10 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
assertEquals(cursor.getMarkDeletedPosition(), p1);
}
- @Test(timeOut = 20000)
- void testSkipEntries() throws Exception {
- ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(2));
+ @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
+ void testSkipEntries(boolean useOpenRangeSet) throws Exception {
+ ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig()
+ .setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet).setMaxEntriesPerLedger(2));
Position pos;
ManagedCursor c1 = ledger.openCursor("c1");
@@ -1368,10 +1376,10 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
assertEquals(c1.getMarkDeletedPosition(), pos);
}
- @Test(timeOut = 20000)
- void testSkipEntriesWithIndividualDeletedMessages() throws Exception {
- ManagedLedger ledger = factory.open("testSkipEntriesWithIndividualDeletedMessages",
- new ManagedLedgerConfig().setMaxEntriesPerLedger(5));
+ @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
+ void testSkipEntriesWithIndividualDeletedMessages(boolean useOpenRangeSet) throws Exception {
+ ManagedLedger ledger = factory.open("testSkipEntriesWithIndividualDeletedMessages", new ManagedLedgerConfig()
+ .setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet).setMaxEntriesPerLedger(5));
ManagedCursor c1 = ledger.openCursor("c1");
Position pos1 = ledger.addEntry("dummy-entry-1".getBytes(Encoding));
@@ -1404,9 +1412,10 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
assertEquals(c1.getMarkDeletedPosition(), pos4);
}
- @Test(timeOut = 20000)
- void testClearBacklog() throws Exception {
- ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig().setMaxEntriesPerLedger(1));
+ @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
+ void testClearBacklog(boolean useOpenRangeSet) throws Exception {
+ ManagedLedger ledger = factory.open("my_test_ledger", new ManagedLedgerConfig()
+ .setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet).setMaxEntriesPerLedger(1));
ManagedCursor c1 = ledger.openCursor("c1");
ledger.addEntry("dummy-entry-1".getBytes(Encoding));
@@ -1455,10 +1464,10 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
factory2.shutdown();
}
- @Test(timeOut = 20000)
- void testRateLimitMarkDelete() throws Exception {
+ @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
+ void testRateLimitMarkDelete(boolean useOpenRangeSet) throws Exception {
ManagedLedgerConfig config = new ManagedLedgerConfig();
- config.setThrottleMarkDelete(1); // Throttle to 1/s
+ config.setThrottleMarkDelete(1).setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet); // Throttle to 1/s
ManagedLedger ledger = factory.open("my_test_ledger", config);
ManagedCursor c1 = ledger.openCursor("c1");
@@ -1484,9 +1493,10 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
factory2.shutdown();
}
- @Test(timeOut = 20000)
- void deleteSingleMessageTwice() throws Exception {
- ManagedLedger ledger = factory.open("my_test_ledger");
+ @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
+ void deleteSingleMessageTwice(boolean useOpenRangeSet) throws Exception {
+ ManagedLedger ledger = factory.open("my_test_ledger",
+ new ManagedLedgerConfig().setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet));
ManagedCursor c1 = ledger.openCursor("c1");
@@ -1551,9 +1561,10 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
assertEquals(c1.getReadPosition(), p4.getNext());
}
- @Test(timeOut = 10000)
- void testReadEntriesOrWait() throws Exception {
- ManagedLedger ledger = factory.open("my_test_ledger");
+ @Test(timeOut = 10000, dataProvider = "useOpenRangeSet")
+ void testReadEntriesOrWait(boolean useOpenRangeSet) throws Exception {
+ ManagedLedger ledger = factory.open("my_test_ledger",
+ new ManagedLedgerConfig().setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet));
final int Consumers = 10;
final CountDownLatch counter = new CountDownLatch(Consumers);
@@ -1846,10 +1857,10 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
p2);
}
- @Test(timeOut = 20000)
- void testFindNewestMatchingEdgeCase6() throws Exception {
- ManagedLedger ledger = factory.open("testFindNewestMatchingEdgeCase6",
- new ManagedLedgerConfig().setMaxEntriesPerLedger(3));
+ @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
+ void testFindNewestMatchingEdgeCase6(boolean useOpenRangeSet) throws Exception {
+ ManagedLedger ledger = factory.open("testFindNewestMatchingEdgeCase6", new ManagedLedgerConfig()
+ .setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet).setMaxEntriesPerLedger(3));
ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
ledger.addEntry("expired".getBytes(Encoding));
@@ -1865,9 +1876,10 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
newPosition);
}
- @Test(timeOut = 20000)
- void testFindNewestMatchingEdgeCase7() throws Exception {
- ManagedLedger ledger = factory.open("testFindNewestMatchingEdgeCase7");
+ @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
+ void testFindNewestMatchingEdgeCase7(boolean useOpenRangeSet) throws Exception {
+ ManagedLedger ledger = factory.open("testFindNewestMatchingEdgeCase7",
+ new ManagedLedgerConfig().setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet));
ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
ledger.addEntry("expired".getBytes(Encoding));
@@ -1959,9 +1971,10 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
lastPosition);
}
- @Test(timeOut = 20000)
- void testIndividuallyDeletedMessages() throws Exception {
- ManagedLedger ledger = factory.open("testIndividuallyDeletedMessages");
+ @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
+ void testIndividuallyDeletedMessages(boolean useOpenRangeSet) throws Exception {
+ ManagedLedger ledger = factory.open("testIndividuallyDeletedMessages",
+ new ManagedLedgerConfig().setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet));
ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
ledger.addEntry("entry-0".getBytes(Encoding));
@@ -1998,9 +2011,10 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
assertTrue(c1.isIndividuallyDeletedEntriesEmpty());
}
- @Test(timeOut = 20000)
- void testIndividuallyDeletedMessages2() throws Exception {
- ManagedLedger ledger = factory.open("testIndividuallyDeletedMessages2");
+ @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
+ void testIndividuallyDeletedMessages2(boolean useOpenRangeSet) throws Exception {
+ ManagedLedger ledger = factory.open("testIndividuallyDeletedMessages2",
+ new ManagedLedgerConfig().setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet));
ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
ledger.addEntry("entry-0".getBytes(Encoding));
@@ -2018,9 +2032,10 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
assertTrue(c1.isIndividuallyDeletedEntriesEmpty());
}
- @Test(timeOut = 20000)
- void testIndividuallyDeletedMessages3() throws Exception {
- ManagedLedger ledger = factory.open("testIndividuallyDeletedMessages3");
+ @Test(timeOut = 20000, dataProvider = "useOpenRangeSet")
+ void testIndividuallyDeletedMessages3(boolean useOpenRangeSet) throws Exception {
+ ManagedLedger ledger = factory.open("testIndividuallyDeletedMessages3",
+ new ManagedLedgerConfig().setUnackedRangesOpenCacheSetEnabled(useOpenRangeSet));
ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
ledger.addEntry("entry-0".getBytes(Encoding));
@@ -2846,5 +2861,44 @@ public class ManagedCursorTest extends MockedBookKeeperTestCase {
assertFalse(cursor.isActive());
}
+ @Test
+ public void deleteMessagesCheckhMarkDelete() throws Exception {
+ ManagedLedger ledger = factory.open("my_test_ledger");
+ ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor("c1");
+ final int totalEntries = 1000;
+ final Position[] positions = new Position[totalEntries];
+ for (int i = 0; i < totalEntries; i++) {
+ // add entry
+ positions[i] = ledger.addEntry(("entry-" + i).getBytes(Encoding));
+ }
+ assertEquals(c1.getNumberOfEntries(), totalEntries);
+ int totalDeletedMessages = 0;
+ for (int i = 0; i < totalEntries; i++) {
+ // delete entry
+ if ((i % 3) == 0) {
+ c1.delete(positions[i]);
+ totalDeletedMessages += 1;
+ }
+ }
+ assertEquals(c1.getNumberOfEntriesInBacklog(), totalEntries - totalDeletedMessages);
+ assertEquals(c1.getNumberOfEntries(), totalEntries - totalDeletedMessages);
+ assertEquals(c1.getMarkDeletedPosition(), positions[0]);
+ assertEquals(c1.getReadPosition(), positions[1]);
+
+ // delete 1/2 of the messags
+ for (int i = 0; i < totalEntries / 2; i++) {
+ // delete entry
+ if ((i % 3) != 0) {
+ c1.delete(positions[i]);
+ totalDeletedMessages += 1;
+ }
+ }
+ int markDelete = totalEntries / 2 - 1;
+ assertEquals(c1.getNumberOfEntriesInBacklog(), totalEntries - totalDeletedMessages);
+ assertEquals(c1.getNumberOfEntries(), totalEntries - totalDeletedMessages);
+ assertEquals(c1.getMarkDeletedPosition(), positions[markDelete]);
+ assertEquals(c1.getReadPosition(), positions[markDelete + 1]);
+ }
+
private static final Logger log = LoggerFactory.getLogger(ManagedCursorTest.class);
}
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryTest.java
index b704db7..8d0c917 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryTest.java
@@ -62,13 +62,13 @@ public class ManagedLedgerFactoryTest extends MockedBookKeeperTestCase {
assertEquals(cursorInfo.markDelete.ledgerId, 3);
assertEquals(cursorInfo.markDelete.entryId, -1);
- assertEquals(cursorInfo.individualDeletedMessages.size(), 1);
+ assertEquals(cursorInfo.individualDeletedMessages.size(), 2);
MessageRangeInfo mri = cursorInfo.individualDeletedMessages.get(0);
- assertEquals(mri.from.ledgerId, p1.getLedgerId());
- assertEquals(mri.from.entryId, p1.getEntryId());
- assertEquals(mri.to.ledgerId, p3.getLedgerId());
- assertEquals(mri.to.entryId, p3.getEntryId());
+ assertEquals(mri.from.ledgerId, p2.getLedgerId());
+ assertEquals(mri.from.entryId, -1);
+ assertEquals(mri.to.ledgerId, p2.getLedgerId());
+ assertEquals(mri.to.entryId, 0);
}
}
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 60d5a9c..984adf1 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -34,7 +34,6 @@ import lombok.Setter;
import org.apache.bookkeeper.client.api.DigestType;
import org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider;
import org.apache.pulsar.common.api.Commands;
-import org.apache.pulsar.common.conf.InternalConfigurationData;
import org.apache.pulsar.common.configuration.Category;
import org.apache.pulsar.common.configuration.FieldContext;
import org.apache.pulsar.common.configuration.PulsarConfiguration;
@@ -885,6 +884,11 @@ public class ServiceConfiguration implements PulsarConfiguration {
+ " unacked ranges into bookkeeper to avoid additional data overhead into zookeeper.")
private int managedLedgerMaxUnackedRangesToPersistInZooKeeper = 1000;
@FieldContext(
+ category = CATEGORY_STORAGE_OFFLOADING,
+ doc = "Use Open Range-Set to cache unacked messages (it is memory efficient but it can take more cpu)"
+ )
+ private boolean managedLedgerUnackedRangesOpenCacheSetEnabled = true;
+ @FieldContext(
dynamic = true,
category = CATEGORY_STORAGE_ML,
doc = "Skip reading non-recoverable/unreadable data-ledger under managed-ledger's list.\n\n"
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index b26a6c1..1890efa 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -750,6 +750,8 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
managedLedgerConfig.setReadEntryTimeoutSeconds(serviceConfig.getManagedLedgerReadEntryTimeoutSeconds());
managedLedgerConfig.setAddEntryTimeoutSeconds(serviceConfig.getManagedLedgerAddEntryTimeoutSeconds());
managedLedgerConfig.setMetadataEnsembleSize(serviceConfig.getManagedLedgerDefaultEnsembleSize());
+ managedLedgerConfig.setUnackedRangesOpenCacheSetEnabled(
+ serviceConfig.isManagedLedgerUnackedRangesOpenCacheSetEnabled());
managedLedgerConfig.setMetadataWriteQuorumSize(serviceConfig.getManagedLedgerDefaultWriteQuorum());
managedLedgerConfig.setMetadataAckQuorumSize(serviceConfig.getManagedLedgerDefaultAckQuorum());
managedLedgerConfig
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageRedeliveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageRedeliveryTest.java
new file mode 100644
index 0000000..819b0c1
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageRedeliveryTest.java
@@ -0,0 +1,221 @@
+/**
+ * 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.pulsar.client.impl;
+
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.assertEquals;
+
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import com.google.common.collect.Sets;
+
+import io.netty.util.concurrent.DefaultThreadFactory;
+
+public class MessageRedeliveryTest extends ProducerConsumerBase {
+ private static final Logger log = LoggerFactory.getLogger(MessageRedeliveryTest.class);
+
+ @BeforeMethod
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ super.producerBaseSetup();
+ }
+
+ @AfterMethod
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @DataProvider(name = "useOpenRangeSet")
+ public static Object[][] useOpenRangeSet() {
+ return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } };
+ }
+
+ /**
+ * It tests that ManagedCursor tracks individually deleted messages and markDeletePosition correctly with different
+ * range-set implementation and re-delivers messages as expected.
+ *
+ * @param useOpenRangeSet
+ * @throws Exception
+ */
+ @Test(dataProvider = "useOpenRangeSet", timeOut = 30000)
+ public void testRedelivery(boolean useOpenRangeSet) throws Exception {
+
+ this.conf.setManagedLedgerMaxEntriesPerLedger(5);
+ this.conf.setManagedLedgerMinLedgerRolloverTimeMinutes(0);
+ this.conf.setManagedLedgerUnackedRangesOpenCacheSetEnabled(useOpenRangeSet);
+ final ScheduledExecutorService executor = Executors.newScheduledThreadPool(20,
+ new DefaultThreadFactory("pulsar"));
+ try {
+ final String ns1 = "my-property/brok-ns1";
+ final String subName = "my-subscriber-name";
+ final int numMessages = 50;
+ admin.namespaces().createNamespace(ns1, Sets.newHashSet("test"));
+
+ final String topic1 = "persistent://" + ns1 + "/my-topic";
+
+ ConsumerImpl<byte[]> consumer1 = (ConsumerImpl<byte[]>) pulsarClient.newConsumer().topic(topic1)
+ .subscriptionName(subName).subscriptionType(SubscriptionType.Shared).receiverQueueSize(10)
+ .acknowledgmentGroupTime(0, TimeUnit.MILLISECONDS).subscribe();
+ ConsumerImpl<byte[]> consumer2 = (ConsumerImpl<byte[]>) pulsarClient.newConsumer().topic(topic1)
+ .subscriptionName(subName).subscriptionType(SubscriptionType.Shared).receiverQueueSize(10)
+ .acknowledgmentGroupTime(0, TimeUnit.MILLISECONDS).subscribe();
+ ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer().topic(topic1).create();
+
+ for (int i = 0; i < numMessages; i++) {
+ String message = "my-message-" + i;
+ producer.send(message.getBytes());
+ }
+
+ CountDownLatch latch = new CountDownLatch(numMessages);
+ AtomicBoolean consume1 = new AtomicBoolean(true);
+ AtomicBoolean consume2 = new AtomicBoolean(true);
+ Set<String> ackedMessages = Sets.newConcurrentHashSet();
+ AtomicInteger counter = new AtomicInteger(0);
+
+ // (1) ack alternate message from consumer-1 which creates ack-hole.
+ executor.submit(() -> {
+ while (true) {
+ try {
+ Message<byte[]> msg = consumer1.receive(1000, TimeUnit.MILLISECONDS);
+ if (msg != null) {
+ if (counter.getAndIncrement() % 2 == 0) {
+ try {
+ consumer1.acknowledge(msg);
+ // ack alternate messages
+ ackedMessages.add(new String(msg.getData()));
+ } catch (PulsarClientException e1) {
+ log.warn("Failed to ack message {}", e1.getMessage());
+ }
+ }
+ } else {
+ break;
+ }
+ } catch (PulsarClientException e2) {
+ // Ok
+ break;
+ }
+ latch.countDown();
+ }
+ });
+
+ // (2) ack all the consumed messages from consumer-2
+ executor.submit(() -> {
+ while (consume2.get()) {
+ try {
+ Message<byte[]> msg = consumer2.receive(1000, TimeUnit.MILLISECONDS);
+ if (msg != null) {
+ consumer2.acknowledge(msg);
+ // ack alternate messages
+ ackedMessages.add(new String(msg.getData()));
+ } else {
+ break;
+ }
+ } catch (PulsarClientException e2) {
+ // Ok
+ break;
+ }
+ latch.countDown();
+ }
+ });
+
+ latch.await(10000, TimeUnit.MILLISECONDS);
+ consume1.set(false);
+
+ // (3) sleep so, consumer2 should timeout on it's pending read operation and not consume more messages
+ Thread.sleep(1000);
+
+ // (4) here we consume all messages but consumer1 only acked alternate messages.
+ assertNotEquals(ackedMessages.size(), numMessages);
+
+ PersistentTopic pTopic = (PersistentTopic) this.pulsar.getBrokerService().getTopicIfExists(topic1).get()
+ .get();
+ ManagedLedgerImpl ml = (ManagedLedgerImpl) pTopic.getManagedLedger();
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ml.getCursors().iterator().next();
+
+ // (5) now, close consumer1 and let broker deliver consumer1's unack messages to consumer2
+ consumer1.close();
+
+ // (6) broker should redeliver all unack messages of consumer-1 and consumer-2 should ack all of them
+ CountDownLatch latch2 = new CountDownLatch(1);
+ executor.submit(() -> {
+ while (true) {
+ try {
+ Message<byte[]> msg = consumer2.receive(1000, TimeUnit.MILLISECONDS);
+ if (msg != null) {
+ consumer2.acknowledge(msg);
+ // ack alternate messages
+ ackedMessages.add(new String(msg.getData()));
+ } else {
+ break;
+ }
+ } catch (PulsarClientException e2) {
+ // Ok
+ break;
+ }
+ if (ackedMessages.size() == numMessages)
+ latch2.countDown();
+ }
+
+ });
+
+ latch2.await(20000, TimeUnit.MILLISECONDS);
+
+ consumer2.close();
+
+ assertEquals(ackedMessages.size(), numMessages);
+
+ // (7) acked message set should be empty
+ assertEquals(cursor.getIndividuallyDeletedMessagesSet().size(), 0);
+
+ // markDelete position should be one position behind read position
+ assertEquals(cursor.getReadPosition(), cursor.getMarkDeletedPosition().getNext());
+
+ producer.close();
+ consumer2.close();
+ } finally {
+ executor.shutdown();
+ }
+
+ }
+
+}