Skip to content

Commit

Permalink
[fix][broker] Fix cursor should use latest ledger config (apache#22644)
Browse files Browse the repository at this point in the history 
Signed-off-by: Zixuan Liu <nodeces@gmail.com>
(cherry picked from commit 774a5d4)
(cherry picked from commit 919cfeb)
  • Loading branch information
@nodece @nikhil-ctds
nodece authored and nikhil-ctds committed Jun 7, 2024
1 parent 493657b commit 1ef7410
Show file tree
Hide file tree
Showing 13 changed files with 76 additions and 56 deletions.
59 changes: 29 additions & 30 deletions managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -119,7 +119,6 @@ public class ManagedCursorImpl implements ManagedCursor {
return 0;
};
protected final BookKeeper bookkeeper;
protected final ManagedLedgerConfig config;
protected final ManagedLedgerImpl ledger;
private final String name;

Expand Down Expand Up @@ -299,31 +298,30 @@ public interface VoidCallback {
void operationFailed(ManagedLedgerException exception);
}

ManagedCursorImpl(BookKeeper bookkeeper, ManagedLedgerConfig config, ManagedLedgerImpl ledger, String cursorName) {
ManagedCursorImpl(BookKeeper bookkeeper, ManagedLedgerImpl ledger, String cursorName) {
this.bookkeeper = bookkeeper;
this.cursorProperties = Collections.emptyMap();
this.config = config;
this.ledger = ledger;
this.name = cursorName;
this.individualDeletedMessages = new RangeSetWrapper<>(positionRangeConverter,
positionRangeReverseConverter, this);
if (config.isDeletionAtBatchIndexLevelEnabled()) {
if (getConfig().isDeletionAtBatchIndexLevelEnabled()) {
this.batchDeletedIndexes = new ConcurrentSkipListMap<>();
} else {
this.batchDeletedIndexes = null;
}
this.digestType = BookKeeper.DigestType.fromApiDigestType(config.getDigestType());
this.digestType = BookKeeper.DigestType.fromApiDigestType(getConfig().getDigestType());
STATE_UPDATER.set(this, State.Uninitialized);
PENDING_MARK_DELETED_SUBMITTED_COUNT_UPDATER.set(this, 0);
PENDING_READ_OPS_UPDATER.set(this, 0);
RESET_CURSOR_IN_PROGRESS_UPDATER.set(this, FALSE);
WAITING_READ_OP_UPDATER.set(this, null);
this.clock = config.getClock();
this.clock = getConfig().getClock();
this.lastActive = this.clock.millis();
this.lastLedgerSwitchTimestamp = this.clock.millis();

if (config.getThrottleMarkDelete() > 0.0) {
markDeleteLimiter = RateLimiter.create(config.getThrottleMarkDelete());
if (getConfig().getThrottleMarkDelete() > 0.0) {
markDeleteLimiter = RateLimiter.create(getConfig().getThrottleMarkDelete());
} else {
// Disable mark-delete rate limiter
markDeleteLimiter = null;
Expand Down Expand Up @@ -602,7 +600,7 @@ protected void recoverFromLedger(final ManagedCursorInfo info, final VoidCallbac
if (positionInfo.getIndividualDeletedMessagesCount() > 0) {
recoverIndividualDeletedMessages(positionInfo.getIndividualDeletedMessagesList());
}
if (config.isDeletionAtBatchIndexLevelEnabled()
if (getConfig().isDeletionAtBatchIndexLevelEnabled()
&& positionInfo.getBatchedEntryDeletionIndexInfoCount() > 0) {
recoverBatchDeletedIndexes(positionInfo.getBatchedEntryDeletionIndexInfoList());
}
Expand All @@ -611,7 +609,8 @@ protected void recoverFromLedger(final ManagedCursorInfo info, final VoidCallbac
}, null);
};
try {
bookkeeper.asyncOpenLedger(ledgerId, digestType, config.getPassword(), openCallback, null);
bookkeeper.asyncOpenLedger(ledgerId, digestType, getConfig().getPassword(), openCallback,
null);
} catch (Throwable t) {
log.error("[{}] Encountered error on opening cursor ledger {} for cursor {}",
ledger.getName(), ledgerId, name, t);
Expand Down Expand Up @@ -968,10 +967,10 @@ public void asyncReadEntriesWithSkipOrWait(int maxEntries, long maxSizeBytes, Re

// Check again for new entries after the configured time, then if still no entries are available register
// to be notified
if (config.getNewEntriesCheckDelayInMillis() > 0) {
if (getConfig().getNewEntriesCheckDelayInMillis() > 0) {
ledger.getScheduledExecutor()
.schedule(() -> checkForNewEntries(op, callback, ctx),
config.getNewEntriesCheckDelayInMillis(), TimeUnit.MILLISECONDS);
getConfig().getNewEntriesCheckDelayInMillis(), TimeUnit.MILLISECONDS);
} else {
// If there's no delay, check directly from the same thread
checkForNewEntries(op, callback, ctx);
Expand Down Expand Up @@ -1319,7 +1318,7 @@ public void operationComplete() {
lastMarkDeleteEntry = new MarkDeleteEntry(newMarkDeletePosition, isCompactionCursor()
? getProperties() : Collections.emptyMap(), null, null);
individualDeletedMessages.clear();
if (config.isDeletionAtBatchIndexLevelEnabled()) {
if (getConfig().isDeletionAtBatchIndexLevelEnabled()) {
batchDeletedIndexes.values().forEach(BitSetRecyclable::recycle);
batchDeletedIndexes.clear();
long[] resetWords = newReadPosition.ackSet;
Expand Down Expand Up @@ -1578,7 +1577,7 @@ protected long getNumberOfEntries(Range<PositionImpl> range) {

lock.readLock().lock();
try {
if (config.isUnackedRangesOpenCacheSetEnabled()) {
if (getConfig().isUnackedRangesOpenCacheSetEnabled()) {
int cardinality = individualDeletedMessages.cardinality(
range.lowerEndpoint().ledgerId, range.lowerEndpoint().entryId,
range.upperEndpoint().ledgerId, range.upperEndpoint().entryId);
Expand Down Expand Up @@ -1958,7 +1957,7 @@ public void asyncMarkDelete(final Position position, Map<String, Long> propertie

PositionImpl newPosition = (PositionImpl) position;

if (config.isDeletionAtBatchIndexLevelEnabled()) {
if (getConfig().isDeletionAtBatchIndexLevelEnabled()) {
if (newPosition.ackSet != null) {
AtomicReference<BitSetRecyclable> bitSetRecyclable = new AtomicReference<>();
BitSetRecyclable givenBitSet = BitSetRecyclable.create().resetWords(newPosition.ackSet);
Expand Down Expand Up @@ -2141,7 +2140,7 @@ public void operationComplete() {
try {
individualDeletedMessages.removeAtMost(mdEntry.newPosition.getLedgerId(),
mdEntry.newPosition.getEntryId());
if (config.isDeletionAtBatchIndexLevelEnabled()) {
if (getConfig().isDeletionAtBatchIndexLevelEnabled()) {
Map<PositionImpl, BitSetRecyclable> subMap = batchDeletedIndexes.subMap(PositionImpl.EARLIEST,
false, PositionImpl.get(mdEntry.newPosition.getLedgerId(),
mdEntry.newPosition.getEntryId()), true);
Expand Down Expand Up @@ -2279,7 +2278,7 @@ public void asyncDelete(Iterable<Position> positions, AsyncCallbacks.DeleteCallb
}

if (isMessageDeleted(position)) {
if (config.isDeletionAtBatchIndexLevelEnabled()) {
if (getConfig().isDeletionAtBatchIndexLevelEnabled()) {
BitSetRecyclable bitSetRecyclable = batchDeletedIndexes.remove(position);
if (bitSetRecyclable != null) {
bitSetRecyclable.recycle();
Expand All @@ -2291,7 +2290,7 @@ public void asyncDelete(Iterable<Position> positions, AsyncCallbacks.DeleteCallb
continue;
}
if (position.ackSet == null) {
if (config.isDeletionAtBatchIndexLevelEnabled()) {
if (getConfig().isDeletionAtBatchIndexLevelEnabled()) {
BitSetRecyclable bitSetRecyclable = batchDeletedIndexes.remove(position);
if (bitSetRecyclable != null) {
bitSetRecyclable.recycle();
Expand All @@ -2308,7 +2307,7 @@ public void asyncDelete(Iterable<Position> positions, AsyncCallbacks.DeleteCallb
log.debug("[{}] [{}] Individually deleted messages: {}", ledger.getName(), name,
individualDeletedMessages);
}
} else if (config.isDeletionAtBatchIndexLevelEnabled()) {
} else if (getConfig().isDeletionAtBatchIndexLevelEnabled()) {
BitSetRecyclable givenBitSet = BitSetRecyclable.create().resetWords(position.ackSet);
BitSetRecyclable bitSet = batchDeletedIndexes.computeIfAbsent(position, (v) -> givenBitSet);
if (givenBitSet != bitSet) {
Expand Down Expand Up @@ -2655,8 +2654,8 @@ public void operationFailed(MetaStoreException e) {
private boolean shouldPersistUnackRangesToLedger() {
return cursorLedger != null
&& !isCursorLedgerReadOnly
&& config.getMaxUnackedRangesToPersist() > 0
&& individualDeletedMessages.size() > config.getMaxUnackedRangesToPersistInMetadataStore();
&& getConfig().getMaxUnackedRangesToPersist() > 0
&& individualDeletedMessages.size() > getConfig().getMaxUnackedRangesToPersistInMetadataStore();
}

private void persistPositionMetaStore(long cursorsLedgerId, PositionImpl position, Map<String, Long> properties,
Expand All @@ -2681,7 +2680,7 @@ private void persistPositionMetaStore(long cursorsLedgerId, PositionImpl positio
info.addAllCursorProperties(buildStringPropertiesMap(cursorProperties));
if (persistIndividualDeletedMessageRanges) {
info.addAllIndividualDeletedMessages(buildIndividualDeletedMessageRanges());
if (config.isDeletionAtBatchIndexLevelEnabled()) {
if (getConfig().isDeletionAtBatchIndexLevelEnabled()) {
info.addAllBatchedEntryDeletionIndexInfo(buildBatchEntryDeletionIndexInfoList());
}
}
Expand Down Expand Up @@ -2946,7 +2945,7 @@ public void operationFailed(ManagedLedgerException exception) {

private CompletableFuture<LedgerHandle> doCreateNewMetadataLedger() {
CompletableFuture<LedgerHandle> future = new CompletableFuture<>();
ledger.asyncCreateLedger(bookkeeper, config, digestType, (rc, lh, ctx) -> {
ledger.asyncCreateLedger(bookkeeper, getConfig(), digestType, (rc, lh, ctx) -> {

if (ledger.checkAndCompleteLedgerOpTask(rc, lh, ctx)) {
future.complete(null);
Expand Down Expand Up @@ -3051,7 +3050,7 @@ private List<MLDataFormats.MessageRange> buildIndividualDeletedMessageRanges() {
acksSerializedSize.addAndGet(messageRange.getSerializedSize());
rangeList.add(messageRange);

return rangeList.size() <= config.getMaxUnackedRangesToPersist();
return rangeList.size() <= getConfig().getMaxUnackedRangesToPersist();
});

this.individualDeletedMessagesSerializedSize = acksSerializedSize.get();
Expand All @@ -3065,7 +3064,7 @@ private List<MLDataFormats.MessageRange> buildIndividualDeletedMessageRanges() {
private List<MLDataFormats.BatchedEntryDeletionIndexInfo> buildBatchEntryDeletionIndexInfoList() {
lock.readLock().lock();
try {
if (!config.isDeletionAtBatchIndexLevelEnabled() || batchDeletedIndexes.isEmpty()) {
if (!getConfig().isDeletionAtBatchIndexLevelEnabled() || batchDeletedIndexes.isEmpty()) {
return Collections.emptyList();
}
MLDataFormats.NestedPositionInfo.Builder nestedPositionBuilder = MLDataFormats.NestedPositionInfo
Expand All @@ -3074,7 +3073,7 @@ private List<MLDataFormats.BatchedEntryDeletionIndexInfo> buildBatchEntryDeletio
.BatchedEntryDeletionIndexInfo.newBuilder();
List<MLDataFormats.BatchedEntryDeletionIndexInfo> result = new ArrayList<>();
Iterator<Map.Entry<PositionImpl, BitSetRecyclable>> iterator = batchDeletedIndexes.entrySet().iterator();
while (iterator.hasNext() && result.size() < config.getMaxBatchDeletedIndexToPersist()) {
while (iterator.hasNext() && result.size() < getConfig().getMaxBatchDeletedIndexToPersist()) {
Map.Entry<PositionImpl, BitSetRecyclable> entry = iterator.next();
nestedPositionBuilder.setLedgerId(entry.getKey().getLedgerId());
nestedPositionBuilder.setEntryId(entry.getKey().getEntryId());
Expand Down Expand Up @@ -3194,8 +3193,8 @@ public void operationFailed(MetaStoreException e) {
boolean shouldCloseLedger(LedgerHandle lh) {
long now = clock.millis();
if (ledger.getFactory().isMetadataServiceAvailable()
&& (lh.getLastAddConfirmed() >= config.getMetadataMaxEntriesPerLedger()
|| lastLedgerSwitchTimestamp < (now - config.getLedgerRolloverTimeout() * 1000))
&& (lh.getLastAddConfirmed() >= getConfig().getMetadataMaxEntriesPerLedger()
|| lastLedgerSwitchTimestamp < (now - getConfig().getLedgerRolloverTimeout() * 1000))
&& (STATE_UPDATER.get(this) != State.Closed && STATE_UPDATER.get(this) != State.Closing)) {
// It's safe to modify the timestamp since this method will be only called from a callback, implying that
// calls will be serialized on one single thread
Expand Down Expand Up @@ -3551,7 +3550,7 @@ private ManagedCursorImpl cursorImpl() {

@Override
public long[] getDeletedBatchIndexesAsLongArray(PositionImpl position) {
if (config.isDeletionAtBatchIndexLevelEnabled()) {
if (getConfig().isDeletionAtBatchIndexLevelEnabled()) {
BitSetRecyclable bitSet = batchDeletedIndexes.get(position);
return bitSet == null ? null : bitSet.toLongArray();
} else {
Expand Down Expand Up @@ -3652,7 +3651,7 @@ public boolean isCacheReadEntry() {
private static final Logger log = LoggerFactory.getLogger(ManagedCursorImpl.class);

public ManagedLedgerConfig getConfig() {
return config;
return getManagedLedger().getConfig();
}

/***
Expand Down
3 changes: 2 additions & 1 deletion managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorMXBeanImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -90,7 +90,8 @@ public long getPersistZookeeperErrors() {

@Override
public void addWriteCursorLedgerSize(final long size) {
writeCursorLedgerSize.add(size * ((ManagedCursorImpl) managedCursor).config.getWriteQuorumSize());
writeCursorLedgerSize.add(
size * managedCursor.getManagedLedger().getConfig().getWriteQuorumSize());
writeCursorLedgerLogicalSize.add(size);
}

Expand Down
8 changes: 4 additions & 4 deletions managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -585,7 +585,7 @@ public void operationComplete(List<String> consumers, Stat s) {
for (final String cursorName : consumers) {
log.info("[{}] Loading cursor {}", name, cursorName);
final ManagedCursorImpl cursor;
cursor = new ManagedCursorImpl(bookKeeper, config, ManagedLedgerImpl.this, cursorName);
cursor = new ManagedCursorImpl(bookKeeper, ManagedLedgerImpl.this, cursorName);

cursor.recover(new VoidCallback() {
@Override
Expand Down Expand Up @@ -616,7 +616,7 @@ public void operationFailed(ManagedLedgerException exception) {
log.debug("[{}] Recovering cursor {} lazily", name, cursorName);
}
final ManagedCursorImpl cursor;
cursor = new ManagedCursorImpl(bookKeeper, config, ManagedLedgerImpl.this, cursorName);
cursor = new ManagedCursorImpl(bookKeeper, ManagedLedgerImpl.this, cursorName);
CompletableFuture<ManagedCursor> cursorRecoveryFuture = new CompletableFuture<>();
uninitializedCursors.put(cursorName, cursorRecoveryFuture);

Expand Down Expand Up @@ -1001,7 +1001,7 @@ public synchronized void asyncOpenCursor(final String cursorName, final InitialP
if (log.isDebugEnabled()) {
log.debug("[{}] Creating new cursor: {}", name, cursorName);
}
final ManagedCursorImpl cursor = new ManagedCursorImpl(bookKeeper, config, this, cursorName);
final ManagedCursorImpl cursor = new ManagedCursorImpl(bookKeeper, this, cursorName);
CompletableFuture<ManagedCursor> cursorFuture = new CompletableFuture<>();
uninitializedCursors.put(cursorName, cursorFuture);
PositionImpl position = InitialPosition.Earliest == initialPosition ? getFirstPosition() : getLastPosition();
Expand Down Expand Up @@ -1134,7 +1134,7 @@ public ManagedCursor newNonDurableCursor(Position startCursorPosition, String cu
return cachedCursor;
}

NonDurableCursorImpl cursor = new NonDurableCursorImpl(bookKeeper, config, this, cursorName,
NonDurableCursorImpl cursor = new NonDurableCursorImpl(bookKeeper, this, cursorName,
(PositionImpl) startCursorPosition, initialPosition, isReadCompacted);
cursor.setActive();

Expand Down
5 changes: 2 additions & 3 deletions managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@
import org.apache.bookkeeper.mledger.AsyncCallbacks.CloseCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCursorCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.MarkDeleteCallback;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.common.api.proto.CommandSubscribe;
import org.slf4j.Logger;
Expand All @@ -35,10 +34,10 @@ public class NonDurableCursorImpl extends ManagedCursorImpl {

private final boolean readCompacted;

NonDurableCursorImpl(BookKeeper bookkeeper, ManagedLedgerConfig config, ManagedLedgerImpl ledger, String cursorName,
NonDurableCursorImpl(BookKeeper bookkeeper, ManagedLedgerImpl ledger, String cursorName,
PositionImpl startCursorPosition, CommandSubscribe.InitialPosition initialPosition,
boolean isReadCompacted) {
super(bookkeeper, config, ledger, cursorName);
super(bookkeeper, ledger, cursorName);
this.readCompacted = isReadCompacted;

// Compare with "latest" position marker by using only the ledger id. Since the C++ client is using 48bits to
Expand Down
3 changes: 2 additions & 1 deletion managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,8 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
callback.readEntriesComplete(entries, ctx);
recycle();
});
} else if (cursor.config.isAutoSkipNonRecoverableData() && exception instanceof NonRecoverableLedgerException) {
} else if (cursor.getConfig().isAutoSkipNonRecoverableData()
&& exception instanceof NonRecoverableLedgerException) {
log.warn("[{}][{}] read failed from ledger at position:{} : {}", cursor.ledger.getName(), cursor.getName(),
readPosition, exception.getMessage());
final ManagedLedgerImpl ledger = (ManagedLedgerImpl) cursor.getManagedLedger();
Expand Down
2 changes: 1 addition & 1 deletion managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/RangeSetWrapper.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ public RangeSetWrapper(LongPairConsumer<T> rangeConverter,
RangeBoundConsumer<T> rangeBoundConsumer,
ManagedCursorImpl managedCursor) {
requireNonNull(managedCursor);
this.config = managedCursor.getConfig();
this.config = managedCursor.getManagedLedger().getConfig();
this.rangeConverter = rangeConverter;
this.rangeSet = config.isUnackedRangesOpenCacheSetEnabled()
? new ConcurrentOpenLongPairRangeSet<>(4096, rangeConverter)
Expand Down
5 changes: 2 additions & 3 deletions managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ReadOnlyCursorImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.ReadOnlyCursor;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.PositionBound;
Expand All @@ -31,9 +30,9 @@
@Slf4j
public class ReadOnlyCursorImpl extends ManagedCursorImpl implements ReadOnlyCursor {

public ReadOnlyCursorImpl(BookKeeper bookkeeper, ManagedLedgerConfig config, ManagedLedgerImpl ledger,
public ReadOnlyCursorImpl(BookKeeper bookkeeper, ManagedLedgerImpl ledger,
PositionImpl startPosition, String cursorName) {
super(bookkeeper, config, ledger, cursorName);
super(bookkeeper, ledger, cursorName);

if (startPosition.equals(PositionImpl.EARLIEST)) {
readPosition = ledger.getFirstPosition().getNext();
Expand Down
Loading

0 comments on commit 1ef7410

Please sign in to comment.