[ML] Fix race condition in getManagedLedgerInternalStats when includeLedgerMetadata=true
getLedgerMetadata is an asynchronous operation and the final result shouldn't complete before the
metadata for all ledgers has been retrieved
@@ -37,6 +37,7 @@ import io.netty.util.Recycler.Handle;
import io.netty.util.ReferenceCountUtil;
import java.io.IOException;
import java.time.Clock;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -4045,23 +4046,6 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
stats.lastConfirmedEntry = this.getLastConfirmedEntry().toString();
stats.state = this.getState().toString();
- stats.ledgers = Lists.newArrayList();
- this.getLedgersInfo().forEach((id, li) -> {
- ManagedLedgerInternalStats.LedgerInfo info = new ManagedLedgerInternalStats.LedgerInfo();
- info.ledgerId = li.getLedgerId();
- info.entries = li.getEntries();
- info.size = li.getSize();
- info.offloaded = li.hasOffloadContext() && li.getOffloadContext().getComplete();
- stats.ledgers.add(info);
- if (includeLedgerMetadata) {
- this.getLedgerMetadata(li.getLedgerId()).handle((lMetadata, ex) -> {
- if (ex == null) {
- info.metadata = lMetadata;
- }
- return null;
- });
- }
-
stats.cursors = Maps.newTreeMap();
this.getCursors().forEach(c -> {
ManagedCursorImpl cursor = (ManagedCursorImpl) c;
@@ -4076,14 +4060,46 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
cs.individuallyDeletedMessages = cursor.getIndividuallyDeletedMessages();
cs.lastLedgerSwitchTimestamp = DateFormatter.format(cursor.getLastLedgerSwitchTimestamp());
cs.state = cursor.getState();
- cs.numberOfEntriesSinceFirstNotAckedMessage =
- cursor.getNumberOfEntriesSinceFirstNotAckedMessage();
+ cs.numberOfEntriesSinceFirstNotAckedMessage = cursor.getNumberOfEntriesSinceFirstNotAckedMessage();
cs.totalNonContiguousDeletedMessagesRange = cursor.getTotalNonContiguousDeletedMessagesRange();
cs.properties = cursor.getProperties();
stats.cursors.put(cursor.getName(), cs);
});
+
+ // make a snapshot of the ledgers infos since we are iterating it twice when metadata is included
+ // a list is sufficient since there's no need to lookup by the ledger id
+ List<LedgerInfo> ledgersInfos = new ArrayList<>(this.getLedgersInfo().values());
+
+ // add asynchronous metadata retrieval operations to a hashmap
+ Map<Long, CompletableFuture<String>> ledgerMetadataFutures = new HashMap();
+ if (includeLedgerMetadata) {
+ ledgersInfos.forEach(li -> {
+ long ledgerId = li.getLedgerId();
+ ledgerMetadataFutures.put(ledgerId, this.getLedgerMetadata(ledgerId).exceptionally(throwable -> {
+ log.warn("Getting metadata for ledger {} failed.", ledgerId, throwable);
+ return null;
+ }));
+ });
+ }
+
+ // wait until metadata has been retrieved
+ FutureUtil.waitForAll(ledgerMetadataFutures.values()).thenAccept(__ -> {
+ stats.ledgers = Lists.newArrayList();
+ ledgersInfos.forEach(li -> {
+ ManagedLedgerInternalStats.LedgerInfo info = new ManagedLedgerInternalStats.LedgerInfo();
+ info.ledgerId = li.getLedgerId();
+ info.entries = li.getEntries();
+ info.size = li.getSize();
+ info.offloaded = li.hasOffloadContext() && li.getOffloadContext().getComplete();
+ if (includeLedgerMetadata) {
+ // lookup metadata from the hashmap which contains completed async operations
+ info.metadata = ledgerMetadataFutures.get(li.getLedgerId()).getNow(null);
+ }
+ stats.ledgers.add(info);
});
statFuture.complete(stats);
+ });
+
return statFuture;
}
