Radiancebobo commented on code in PR #4873:
URL: https://github.com/apache/bookkeeper/pull/4873#discussion_r4014283988
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/AuditorTask.java:
##########
@@ -88,24 +88,47 @@ protected CompletableFuture<?>
publishSuspectedLedgersAsync(Collection<String> m
.attr("ledgers", ledgers)
.attr("missingBookies", missingBookies)
.log("Following ledgers are identified as underreplicated");
-
auditorStats.getNumUnderReplicatedLedger().registerSuccessfulValue(ledgers.size());
+ LongAdder publishedLedgers = new LongAdder();
LongAdder underReplicatedSize = new LongAdder();
- FutureUtils.processList(
+ CompletableFuture<List<Void>> publishFuture = FutureUtils.processList(
Lists.newArrayList(ledgers),
- ledgerId ->
-
ledgerManager.readLedgerMetadata(ledgerId).whenComplete((metadata, exception)
-> {
- if (exception == null) {
-
underReplicatedSize.add(metadata.getValue().getLength());
- }
- }), null).whenComplete((res, e) -> {
+ ledgerId ->
ledgerManager.readLedgerMetadata(ledgerId).handle((metadata, exception) -> {
+ if (exception != null) {
+ if (BKException.getExceptionCode(exception)
+ ==
BKException.Code.NoSuchLedgerExistsOnMetadataServerException) {
+ log.info()
+ .attr("ledgerId", ledgerId)
+ .log("Ledger was deleted before publishing
underreplicated mark");
+ return FutureUtils.Void();
+ }
+ log.warn()
+ .attr("ledgerId", ledgerId)
+ .exception(exception)
+ .log("Unable to read ledger metadata;
publishing underreplicated mark fail-open");
+ } else if (metadata == null || metadata.getValue() ==
null) {
+ log.warn()
+ .attr("ledgerId", ledgerId)
+ .log("Ledger metadata was empty; publishing
underreplicated mark fail-open");
+ } else if (metadata.getValue().getWriteQuorumSize() == 1) {
+ auditorStats.getNumSingleReplicaLedgersSkipped().inc();
+ log.info()
+ .attr("ledgerId", ledgerId)
+ .attr("writeQuorumSize",
metadata.getValue().getWriteQuorumSize())
+ .attr("reason", "single-replica-ledger")
+ .attr("action", "skip-publish")
+ .log("Skipping underreplicated mark");
+ return FutureUtils.Void();
+ } else {
+
underReplicatedSize.add(metadata.getValue().getLength());
+ }
+
+ publishedLedgers.increment();
+ return
ledgerUnderreplicationManager.markLedgerUnderreplicatedAsync(ledgerId,
missingBookies);
+ }).thenCompose(markFuture -> markFuture),
null).whenComplete((res, e) -> {
Review Comment:
`FutureUtils.Void()` returns
`CompletableFuture<Void>`, not `Void`. In addition,
`markLedgerUnderreplicatedAsync(...)` also returns
`CompletableFuture<Void>`. Therefore, all branches of the `handle`
callback consistently return `CompletableFuture<Void>`, and
`thenCompose(markFuture -> markFuture)` safely flattens the result.
The branch also compiles successfully and the replication CI tests pass,
so no raw type or runtime cast is involved here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
