LedgerAllocatorPool implements LedgerAllocator { @Override public Future<LedgerHandle> tryObtain(final Transaction<Object> txn, final Transaction.OpListener<LedgerHandle> listener) { final SimpleLedgerAllocator allocator; synchronized (this) { if (allocatingList.isEmpty()) { return Future.exception(new IOException("No ledger allocator available under " + poolPath + ".")); } else { allocator = allocatingList.removeFirst(); } } final Promise<LedgerHandle> tryObtainPromise = new Promise<LedgerHandle>(); final FutureEventListener<LedgerHandle> tryObtainListener = new FutureEventListener<LedgerHandle>() { @Override public void onSuccess(LedgerHandle lh) { synchronized (LedgerAllocatorPool.this) { obtainMap.put(lh, allocator); reverseObtainMap.put(allocator, lh); tryObtainPromise.setValue(lh); } } @Override public void onFailure(Throwable cause) { try { rescueAllocator(allocator); } catch (IOException ioe) { logger.info("Failed to rescue allocator {}", allocator.allocatePath, ioe); } tryObtainPromise.setException(cause); } }; allocator.tryObtain(txn, new Transaction.OpListener<LedgerHandle>() { @Override public void onCommit(LedgerHandle lh) { confirmObtain(allocator); listener.onCommit(lh); } @Override public void onAbort(Throwable t) { abortObtain(allocator); listener.onAbort(t); } }).addEventListener(tryObtainListener); return tryObtainPromise; } LedgerAllocatorPool(String poolPath, int corePoolSize,
                               DistributedLogConfiguration conf,
                               ZooKeeperClient zkc,
                               BookKeeperClient bkc,
                               ScheduledExecutorService scheduledExecutorService); @Override void start(); @VisibleForTesting synchronized int obtainMapSize(); @Override void allocate(); @Override Future<LedgerHandle> tryObtain(final Transaction<Object> txn,
                                          final Transaction.OpListener<LedgerHandle> listener); @Override Future<Void> asyncClose(); @Override Future<Void> delete();  }
@Test(timeout = 60000) public void testObtainWhenNoAllocator() throws Exception { String allocationPath = "/obtainWhenNoAllocator"; LedgerAllocatorPool pool = new LedgerAllocatorPool(allocationPath, 0, dlConf, zkc, bkc, allocationExecutor); ZKTransaction txn = newTxn(); try { FutureUtils.result(pool.tryObtain(txn, NULL_LISTENER)); fail("Should fail obtain ledger handle if there is no allocator."); } catch (SimpleLedgerAllocator.AllocationException ae) { fail("Should fail obtain ledger handle if there is no allocator."); } catch (IOException ioe) { } Utils.close(pool); }