CompositeLeakyBucket implements LeakyBucket, Startable, Stoppable { @VisibleForTesting protected void refresh() { if(StringUtil.isEmpty(keeperConfig.getMetaServerAddress())) { logger.debug("[refresh]address null, will not update token size"); return; } List<RedisKeeperServer> keepers = keeperContainerService.list(); int total = keepers.size(), ack = 0; for (RedisKeeperServer keeperServer : keepers) { if (keeperServer.getRedisMaster().getMasterState() == MASTER_STATE.REDIS_REPL_CONNECTED) { ack ++; } } try { MetaServerKeeperService.KeeperContainerTokenStatusResponse response = metaServerKeeperService.refreshKeeperContainerTokenStatus( new MetaServerKeeperService.KeeperContainerTokenStatusRequest( keeperContainerMeta, ack, total)); if (response == null) { logger.warn("[refresh][no-response]"); return; } if (response.getTokenSize() > origin.getTotalSize()) { origin.resize(response.getTokenSize()); } } catch (Exception e) { logger.error("[refresh]", e); } } 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 testRefresh() throws InterruptedException { when(keeperConfig.isKeeperRateLimitOpen()).thenReturn(false); leakyBucket.setScheduled(scheduled); leakyBucket.checkKeeperConfigChange(); sleep(110); 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.refresh(); } }); for (int i = 0; i < task; i++) { executors.execute(new Runnable() { @Override public void run() { try { barrier.await(); } catch (Exception ignore) { } if (leakyBucket.tryAcquire()) { counter.incrementAndGet(); } latch.countDown(); } }); } latch.await(1000, TimeUnit.MILLISECONDS); Assert.assertTrue(3 < counter.get()); }