ConsoleNotifycationTask extends AbstractLifecycle implements MetaServerStateChangeHandler, TopElement { @Override public void keeperActiveElected(final String clusterId, final String shardId, final KeeperMeta activeKeeper) { logger.info("[keeperActiveElected][called]{},{},{}",clusterId,shardId,activeKeeper); Command<Void> command = new AbstractCommand<Void>() { @Override public String getName() { return "[keeperActiveElected notify]"; } @Override protected void doExecute() throws Exception { logger.info("[keeperActiveElected][execute]{},{},{}",clusterId,shardId,activeKeeper); consoleService.keeperActiveChanged(dc, clusterId, shardId, activeKeeper); future().setSuccess(); } @Override protected void doReset() { } }; oneThreadTaskExecutor.executeCommand(command); } ConsoleNotifycationTask();  ConsoleNotifycationTask(int retryDelayBase); @Override void keeperActiveElected(final String clusterId, final String shardId, final KeeperMeta activeKeeper); void setConsoleService(ConsoleService consoleService);  }
@Test public void testSuccess() throws Exception{ int times = 100; for(int i=0;i<times;i++){ consoleNotifycationTask.keeperActiveElected("cluster1", "shard1", new KeeperMeta()); sleep(10); verify(consoleService, times(i + 1)).keeperActiveChanged(anyString(), anyString(), anyString(), any(KeeperMeta.class)); } }
@Test(expected = RejectedExecutionException.class) public void testDispose() throws Exception{ consoleNotifycationTask.dispose(); consoleNotifycationTask.keeperActiveElected("cluster1", "shard1", new KeeperMeta()); }