Skip to content

[fix][broker] fix No such ledger exception #16420

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

Merged
Jason918 merged 9 commits into from
Jul 8, 2022
Merged

[fix][broker] fix No such ledger exception #16420

Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
ded6d8f
fix No such ledger exception
lordcheng10 Jul 6, 2022
140f29c
move currentLedgerEntries and currentLedgerSize
lordcheng10 Jul 6, 2022
c698df7
move ledgers.put to operationComplete
lordcheng10 Jul 6, 2022
dc505b7
fix ledgers.put
lordcheng10 Jul 7, 2022
76334b7
use ledgersTmp to write zookkeeper,after sucess,update ldgers
lordcheng10 Jul 7, 2022
07a1645
update updateLedgersListAfterRollover
lordcheng10 Jul 7, 2022
e13e2fc
put lh to ledgersTmp
lordcheng10 Jul 7, 2022
b36da7b
extend buildManagedLedgerInfo
lordcheng10 Jul 7, 2022
1200640
getManagedLedgerInfo(newLedger) after get metadataMutex
lordcheng10 Jul 7, 2022
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
32 changes: 20 additions & 12 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 @@ -1482,18 +1482,18 @@ public synchronized void createComplete(int rc, final LedgerHandle lh, Object ct
lastLedgerCreationFailureTimestamp = clock.millis();
} else {
log.info("[{}] Created new ledger {}", name, lh.getId());
ledgers.put(lh.getId(), LedgerInfo.newBuilder().setLedgerId(lh.getId()).setTimestamp(0).build());
currentLedger = lh;
currentLedgerEntries = 0;
currentLedgerSize = 0;

LedgerInfo newLedger = LedgerInfo.newBuilder().setLedgerId(lh.getId()).setTimestamp(0).build();
final MetaStoreCallback<Void> cb = new MetaStoreCallback<Void>() {
@Override
public void operationComplete(Void v, Stat stat) {
if (log.isDebugEnabled()) {
log.debug("[{}] Updating of ledgers list after create complete. version={}", name, stat);
}
ledgersStat = stat;
ledgers.put(lh.getId(), newLedger);
currentLedger = lh;
currentLedgerEntries = 0;
currentLedgerSize = 0;
metadataMutex.unlock();
updateLedgersIdsComplete(stat);
synchronized (ManagedLedgerImpl.this) {
Expand All @@ -1508,8 +1508,6 @@ public void operationComplete(Void v, Stat stat) {
@Override
public void operationFailed(MetaStoreException e) {
log.warn("[{}] Error updating meta data with the new list of ledgers: {}", name, e.getMessage());
// Remove the ledger, since we failed to update the list
ledgers.remove(lh.getId());
mbean.startDataLedgerDeleteOp();
bookKeeper.asyncDeleteLedger(lh.getId(), (rc1, ctx1) -> {
mbean.endDataLedgerDeleteOp();
Expand Down Expand Up @@ -1545,21 +1543,22 @@ public void operationFailed(MetaStoreException e) {
}
};

updateLedgersListAfterRollover(cb);
updateLedgersListAfterRollover(cb, newLedger);
}
}

private void updateLedgersListAfterRollover(MetaStoreCallback<Void> callback) {
private void updateLedgersListAfterRollover(MetaStoreCallback<Void> callback, LedgerInfo newLedger) {
if (!metadataMutex.tryLock()) {
// Defer update for later
scheduledExecutor.schedule(() -> updateLedgersListAfterRollover(callback), 100, TimeUnit.MILLISECONDS);
scheduledExecutor.schedule(() -> updateLedgersListAfterRollover(callback, newLedger),
100, TimeUnit.MILLISECONDS);
return;
}

if (log.isDebugEnabled()) {
log.debug("[{}] Updating ledgers ids with new ledger. version={}", name, ledgersStat);
}
store.asyncUpdateLedgerIds(name, getManagedLedgerInfo(), ledgersStat, callback);
ManagedLedgerInfo mlInfo = getManagedLedgerInfo(newLedger);
store.asyncUpdateLedgerIds(name, mlInfo, ledgersStat, callback);
}

public synchronized void updateLedgersIdsComplete(Stat stat) {
Expand Down Expand Up @@ -3569,8 +3568,17 @@ private ManagedLedgerInfo getManagedLedgerInfo() {
return buildManagedLedgerInfo(ledgers);
}

private ManagedLedgerInfo getManagedLedgerInfo(LedgerInfo newLedger) {
ManagedLedgerInfo.Builder mlInfo = ManagedLedgerInfo.newBuilder().addAllLedgerInfo(ledgers.values())
.addLedgerInfo(newLedger);
return buildManagedLedgerInfo(mlInfo);
}
private ManagedLedgerInfo buildManagedLedgerInfo(Map<Long, LedgerInfo> ledgers) {
ManagedLedgerInfo.Builder mlInfo = ManagedLedgerInfo.newBuilder().addAllLedgerInfo(ledgers.values());
return buildManagedLedgerInfo(mlInfo);
}

private ManagedLedgerInfo buildManagedLedgerInfo(ManagedLedgerInfo.Builder mlInfo) {
if (state == State.Terminated) {
mlInfo.setTerminatedPosition(NestedPositionInfo.newBuilder().setLedgerId(lastConfirmedEntry.getLedgerId())
.setEntryId(lastConfirmedEntry.getEntryId()));
Expand Down