You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@bookkeeper.apache.org by GitBox <gi...@apache.org> on 2018/08/22 21:27:38 UTC
[GitHub] sijie commented on a change in pull request #1619: Provide async
version of markLedgerUnderreplicated for LedgerUnderreplicationManager
sijie commented on a change in pull request #1619: Provide async version of markLedgerUnderreplicated for LedgerUnderreplicationManager
URL: https://github.com/apache/bookkeeper/pull/1619#discussion_r212116114
##########
File path: bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/Auditor.java
##########
@@ -622,91 +611,54 @@ void checkAllLedgers() throws BKAuditException, BKException,
try {
final LedgerChecker checker = new LedgerChecker(client);
- final AtomicInteger returnCode = new AtomicInteger(BKException.Code.OK);
- final CountDownLatch processDone = new CountDownLatch(1);
+ final CompletableFuture<Void> processFuture = new CompletableFuture<>();
- Processor<Long> checkLedgersProcessor = new Processor<Long>() {
- @Override
- public void process(final Long ledgerId,
- final AsyncCallback.VoidCallback callback) {
- try {
- if (!ledgerUnderreplicationManager.isLedgerReplicationEnabled()) {
- LOG.info("Ledger rereplication has been disabled, aborting periodic check");
- processDone.countDown();
- return;
- }
- } catch (ReplicationException.UnavailableException ue) {
- LOG.error("Underreplication manager unavailable running periodic check", ue);
- processDone.countDown();
+ Processor<Long> checkLedgersProcessor = (ledgerId, callback) -> {
+ try {
+ if (!ledgerUnderreplicationManager.isLedgerReplicationEnabled()) {
+ LOG.info("Ledger rereplication has been disabled, aborting periodic check");
+ FutureUtils.complete(processFuture, null);
return;
}
+ } catch (UnavailableException ue) {
+ LOG.error("Underreplication manager unavailable running periodic check", ue);
+ FutureUtils.complete(processFuture, null);
+ return;
+ }
- // Do not perform blocking calls that involve making ZK calls from within the ZK
- // event thread. Jump to background thread instead to avoid deadlock.
- ForkJoinPool.commonPool().execute(() -> {
- LedgerHandle lh = null;
- try {
- lh = admin.openLedgerNoRecovery(ledgerId);
- checker.checkLedger(lh,
- new ProcessLostFragmentsCb(lh, callback),
- conf.getAuditorLedgerVerificationPercentage());
- // we collect the following stats to get a measure of the
- // distribution of a single ledger within the bk cluster
- // the higher the number of fragments/bookies, the more distributed it is
- numFragmentsPerLedger.registerSuccessfulValue(lh.getNumFragments());
- numBookiesPerLedger.registerSuccessfulValue(lh.getNumBookies());
- numLedgersChecked.inc();
- } catch (BKException.BKNoSuchLedgerExistsException bknsle) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Ledger was deleted before we could check it", bknsle);
- }
- callback.processResult(BKException.Code.OK,
- null, null);
- return;
- } catch (BKException bke) {
- LOG.error("Couldn't open ledger " + ledgerId, bke);
- callback.processResult(BKException.Code.BookieHandleNotAvailableException,
- null, null);
- return;
- } catch (InterruptedException ie) {
- LOG.error("Interrupted opening ledger", ie);
- Thread.currentThread().interrupt();
- callback.processResult(BKException.Code.InterruptedException, null, null);
- return;
- } finally {
- if (lh != null) {
- try {
- lh.close();
- } catch (BKException bke) {
- LOG.warn("Couldn't close ledger " + ledgerId, bke);
- } catch (InterruptedException ie) {
- LOG.warn("Interrupted closing ledger " + ledgerId, ie);
- Thread.currentThread().interrupt();
- }
- }
+ admin.asyncOpenLedgerNoRecovery(ledgerId, (rc, lh, ctx) -> {
+ if (Code.OK == rc) {
+ checker.checkLedger(lh,
+ // the ledger handle will be closed after checkLedger is done.
+ new ProcessLostFragmentsCb(lh, callback),
+ conf.getAuditorLedgerVerificationPercentage());
+ // we collect the following stats to get a measure of the
+ // distribution of a single ledger within the bk cluster
+ // the higher the number of fragments/bookies, the more distributed it is
+ numFragmentsPerLedger.registerSuccessfulValue(lh.getNumFragments());
+ numBookiesPerLedger.registerSuccessfulValue(lh.getNumBookies());
+ numLedgersChecked.inc();
+ } else if (Code.NoSuchLedgerExistsException == rc) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Ledger {} was deleted before we could check it", ledgerId);
}
- });
- }
+ callback.processResult(Code.OK, null, null);
+ } else {
+ LOG.error("Couldn't open ledger {} to check : {}", ledgerId, BKException.getMessage(rc));
+ callback.processResult(rc, null, null);
+ }
+ }, null);
};
ledgerManager.asyncProcessLedgers(checkLedgersProcessor,
- new AsyncCallback.VoidCallback() {
- @Override
- public void processResult(int rc, String s, Object obj) {
- returnCode.set(rc);
- processDone.countDown();
- }
- }, null, BKException.Code.OK, BKException.Code.ReadException);
- try {
- processDone.await();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new BKAuditException(
- "Exception while checking ledgers", e);
- }
- if (returnCode.get() != BKException.Code.OK) {
- throw BKException.create(returnCode.get());
- }
+ (rc, path, ctx) -> {
+ if (Code.OK == rc) {
+ FutureUtils.complete(processFuture, null);
+ } else {
+ FutureUtils.completeExceptionally(processFuture, BKException.create(rc));
+ }
+ }, null, BKException.Code.OK, BKException.Code.ReadException);
+ FutureUtils.result(processFuture, BKException.HANDLER);
Review comment:
processFuture is completed only when finished processing the last ledger, so checkAllLedgers is still sync, which is fine, since checkAllLedgers is executing in Auditor's own executor.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services