CompositeLeakyBucket implements LeakyBucket, Startable, Stoppable { @Override public void resize(int newSize) { if(!closed.get()) { origin.resize(newSize); } } CompositeLeakyBucket(KeeperConfig keeperConfig, MetaServerKeeperService metaServerKeeperService,
                                KeeperContainerService keeperContainerService); @Override boolean tryAcquire(); @Override void release(); @Override void resize(int newSize); @Override int references(); @Override int getTotalSize(); @Override void start(); @Override void stop(); boolean isClosed();  }
@Test public void testResize() throws InterruptedException { AtomicInteger counter = new AtomicInteger(); int task = 3 * 100, newSize = 10; CountDownLatch latch = new CountDownLatch(task); CyclicBarrier barrier = new CyclicBarrier(task + 1); executors.execute(new Runnable() { @Override public void run() { try { barrier.await(); } catch (Exception ignore) { } leakyBucket.resize(newSize); } }); for (int i = 0; i < task; i++) { executors.execute(new Runnable() { @Override public void run() { try { barrier.await(); sleep(10); } catch (Exception ignore) { } if (leakyBucket.tryAcquire()) { counter.incrementAndGet(); } latch.countDown(); } }); } latch.await(1000, TimeUnit.MILLISECONDS); Assert.assertEquals(newSize, counter.get()); }