Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[fix][broker] Fix cursor should use latest ledger config (#22644) #13

Merged
merged 1 commit into from
May 11, 2024
Merged

[fix][broker] Fix cursor should use latest ledger config (#22644) #13

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 36 additions & 32 deletions managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,7 +107,6 @@
public class ManagedCursorImpl implements ManagedCursor {

protected final BookKeeper bookkeeper;
protected final ManagedLedgerConfig config;
protected final ManagedLedgerImpl ledger;
private final String name;
private volatile Map<String, String> cursorProperties;
Expand Down Expand Up @@ -282,32 +281,31 @@ 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 = config.isUnackedRangesOpenCacheSetEnabled()
this.individualDeletedMessages = getConfig().isUnackedRangesOpenCacheSetEnabled()
? new ConcurrentOpenLongPairRangeSet<>(4096, positionRangeConverter)
: new LongPairRangeSet.DefaultRangeSet<>(positionRangeConverter);
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 @@ -537,16 +535,17 @@ protected void recoverFromLedger(final ManagedCursorInfo info, final VoidCallbac
if (positionInfo.getIndividualDeletedMessagesCount() > 0) {
recoverIndividualDeletedMessages(positionInfo.getIndividualDeletedMessagesList());
}
if (config.isDeletionAtBatchIndexLevelEnabled() && batchDeletedIndexes != null
&& positionInfo.getBatchedEntryDeletionIndexInfoCount() > 0) {
if (getConfig().isDeletionAtBatchIndexLevelEnabled() && batchDeletedIndexes != null
&& positionInfo.getBatchedEntryDeletionIndexInfoCount() > 0) {
recoverBatchDeletedIndexes(positionInfo.getBatchedEntryDeletionIndexInfoList());
}
recoveredCursor(position, recoveredProperties, cursorProperties, lh);
callback.operationComplete();
}, 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 @@ -878,10 +877,10 @@ public void asyncReadEntriesOrWait(int maxEntries, long maxSizeBytes, ReadEntrie

// 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 @@ -1201,7 +1200,7 @@ public void operationComplete() {
lastMarkDeleteEntry = new MarkDeleteEntry(newMarkDeletePosition, isCompactionCursor()
? getProperties() : Collections.emptyMap(), null, null);
individualDeletedMessages.clear();
if (config.isDeletionAtBatchIndexLevelEnabled() && batchDeletedIndexes != null) {
if (getConfig().isDeletionAtBatchIndexLevelEnabled() && batchDeletedIndexes != null) {
batchDeletedIndexes.values().forEach(BitSetRecyclable::recycle);
batchDeletedIndexes.clear();
long[] resetWords = newReadPosition.ackSet;
Expand Down Expand Up @@ -1466,7 +1465,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 @@ -1829,7 +1828,7 @@ public void asyncMarkDelete(final Position position, Map<String, Long> propertie

PositionImpl newPosition = (PositionImpl) position;

if (config.isDeletionAtBatchIndexLevelEnabled() && batchDeletedIndexes != null) {
if (getConfig().isDeletionAtBatchIndexLevelEnabled() && batchDeletedIndexes != null) {
if (newPosition.ackSet != null) {
batchDeletedIndexes.put(newPosition, BitSetRecyclable.create().resetWords(newPosition.ackSet));
newPosition = ledger.getPreviousPosition(newPosition);
Expand Down Expand Up @@ -1992,7 +1991,7 @@ public void operationComplete() {
try {
individualDeletedMessages.removeAtMost(mdEntry.newPosition.getLedgerId(),
mdEntry.newPosition.getEntryId());
if (config.isDeletionAtBatchIndexLevelEnabled() && batchDeletedIndexes != null) {
if (getConfig().isDeletionAtBatchIndexLevelEnabled() && batchDeletedIndexes != null) {
Map<PositionImpl, BitSetRecyclable> subMap = batchDeletedIndexes.subMap(PositionImpl.EARLIEST,
false, PositionImpl.get(mdEntry.newPosition.getLedgerId(),
mdEntry.newPosition.getEntryId()), true);
Expand Down Expand Up @@ -2120,8 +2119,8 @@ public void asyncDelete(Iterable<Position> positions, AsyncCallbacks.DeleteCallb
}

if (individualDeletedMessages.contains(position.getLedgerId(), position.getEntryId())
|| position.compareTo(markDeletePosition) <= 0) {
if (config.isDeletionAtBatchIndexLevelEnabled() && batchDeletedIndexes != null) {
|| position.compareTo(markDeletePosition) <= 0) {
if (getConfig().isDeletionAtBatchIndexLevelEnabled() && batchDeletedIndexes != null) {
BitSetRecyclable bitSetRecyclable = batchDeletedIndexes.remove(position);
if (bitSetRecyclable != null) {
bitSetRecyclable.recycle();
Expand All @@ -2133,7 +2132,7 @@ public void asyncDelete(Iterable<Position> positions, AsyncCallbacks.DeleteCallb
continue;
}
if (position.ackSet == null) {
if (config.isDeletionAtBatchIndexLevelEnabled() && batchDeletedIndexes != null) {
if (getConfig().isDeletionAtBatchIndexLevelEnabled() && batchDeletedIndexes != null) {
BitSetRecyclable bitSetRecyclable = batchDeletedIndexes.remove(position);
if (bitSetRecyclable != null) {
bitSetRecyclable.recycle();
Expand All @@ -2150,7 +2149,7 @@ public void asyncDelete(Iterable<Position> positions, AsyncCallbacks.DeleteCallb
log.debug("[{}] [{}] Individually deleted messages: {}", ledger.getName(), name,
individualDeletedMessages);
}
} else if (config.isDeletionAtBatchIndexLevelEnabled() && batchDeletedIndexes != null) {
} else if (getConfig().isDeletionAtBatchIndexLevelEnabled() && batchDeletedIndexes != null) {
BitSetRecyclable givenBitSet = BitSetRecyclable.create().resetWords(position.ackSet);
BitSetRecyclable bitSet = batchDeletedIndexes.computeIfAbsent(position, (v) -> givenBitSet);
if (givenBitSet != bitSet) {
Expand Down Expand Up @@ -2480,8 +2479,8 @@ public void operationFailed(MetaStoreException e) {
private boolean shouldPersistUnackRangesToLedger() {
return cursorLedger != null
&& !isCursorLedgerReadOnly
&& config.getMaxUnackedRangesToPersist() > 0
&& individualDeletedMessages.size() > config.getMaxUnackedRangesToPersistInZk();
&& getConfig().getMaxUnackedRangesToPersist() > 0
&& individualDeletedMessages.size() > getConfig().getMaxUnackedRangesToPersistInZk();
}

private void persistPositionMetaStore(long cursorsLedgerId, PositionImpl position, Map<String, Long> properties,
Expand All @@ -2504,7 +2503,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 @@ -2681,7 +2680,7 @@ void internalFlushPendingMarkDeletes() {
void createNewMetadataLedger(final VoidCallback callback) {
ledger.mbean.startCursorLedgerCreateOp();

ledger.asyncCreateLedger(bookkeeper, config, digestType, (rc, lh, ctx) -> {
ledger.asyncCreateLedger(bookkeeper, getConfig(), digestType, (rc, lh, ctx) -> {

if (ledger.checkAndCompleteLedgerOpTask(rc, lh, ctx)) {
return;
Expand Down Expand Up @@ -2802,7 +2801,8 @@ private List<MLDataFormats.MessageRange> buildIndividualDeletedMessageRanges() {
MessageRange messageRange = messageRangeBuilder.build();
acksSerializedSize.addAndGet(messageRange.getSerializedSize());
rangeList.add(messageRange);
return rangeList.size() <= config.getMaxUnackedRangesToPersist();

return rangeList.size() <= getConfig().getMaxUnackedRangesToPersist();
});
this.individualDeletedMessagesSerializedSize = acksSerializedSize.get();
return rangeList;
Expand All @@ -2814,7 +2814,7 @@ private List<MLDataFormats.MessageRange> buildIndividualDeletedMessageRanges() {
private List<MLDataFormats.BatchedEntryDeletionIndexInfo> buildBatchEntryDeletionIndexInfoList() {
lock.readLock().lock();
try {
if (!config.isDeletionAtBatchIndexLevelEnabled() || batchDeletedIndexes == null
if (!getConfig().isDeletionAtBatchIndexLevelEnabled() || batchDeletedIndexes == null
|| batchDeletedIndexes.isEmpty()) {
return Collections.emptyList();
}
Expand All @@ -2824,7 +2824,7 @@ private List<MLDataFormats.BatchedEntryDeletionIndexInfo> buildBatchEntryDeletio
.BatchedEntryDeletionIndexInfo.newBuilder();
List<MLDataFormats.BatchedEntryDeletionIndexInfo> result = Lists.newArrayList();
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 @@ -2913,8 +2913,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 @@ -3271,7 +3271,7 @@ private ManagedCursorImpl cursorImpl() {

@Override
public long[] getDeletedBatchIndexesAsLongArray(PositionImpl position) {
if (config.isDeletionAtBatchIndexLevelEnabled() && batchDeletedIndexes != null) {
if (getConfig().isDeletionAtBatchIndexLevelEnabled() && batchDeletedIndexes != null) {
BitSetRecyclable bitSet = batchDeletedIndexes.get(position);
return bitSet == null ? null : bitSet.toLongArray();
} else {
Expand Down Expand Up @@ -3355,5 +3355,9 @@ public void setState(State state) {
this.state = state;
}

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

private static final Logger log = LoggerFactory.getLogger(ManagedCursorImpl.class);
}
3 changes: 2 additions & 1 deletion managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorMXBeanImpl.java
View file
Open in desktop
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
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -554,7 +554,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 @@ -585,7 +585,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 @@ -967,7 +967,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 @@ -1101,7 +1101,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
Open in desktop
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
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,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());
// try to find and move to next valid ledger
Expand Down
Loading
Loading