DefaultKeeperStateController extends AbstractLifecycle implements KeeperStateController, TopElement { @Override public void addKeeper(KeeperTransMeta keeperTransMeta) { logger.info("[addKeeper]{}", keeperTransMeta); KeeperContainerService keeperContainerService = getKeeperContainerService(keeperTransMeta); shardExecutor.execute(new Pair<>(keeperTransMeta.getClusterId(), keeperTransMeta.getShardId()), createAddKeeperCommand(keeperContainerService, keeperTransMeta, scheduled, addKeeperSuccessTimeoutMilli)); }  @Override void addKeeper(KeeperTransMeta keeperTransMeta); @Override void removeKeeper(KeeperTransMeta keeperTransMeta); void setExecutors(ExecutorService executors);  }
@Test public void testAdd(){ addCommandSupplier = new Supplier<Command<?>>() { @Override public Command<?> get() { return addCommand; } }; Assert.assertFalse(addCommand.isBeginExecute()); defaultKeeperStateController.addKeeper(new KeeperTransMeta(getClusterId(), getShardId(), new KeeperMeta())); sleep(50); Assert.assertTrue(addCommand.isBeginExecute()); }
@Test public void testSequentialExecInsideShard() throws Exception { AtomicInteger counter = new AtomicInteger(); addCommandSupplier = new Supplier<Command<?>>() { @Override public Command<?> get() { return new CountingCommand(counter, 100); } }; int tasks = 100; for(int i = 0; i < tasks; i++) { defaultKeeperStateController.addKeeper(new KeeperTransMeta(getClusterId(), getShardId(), new KeeperMeta())); } sleep(250); Assert.assertTrue(counter.get() < tasks && counter.get() > 0); }
@Test public void testSequentialExecBetweenShards() throws Exception { AtomicInteger counter = new AtomicInteger(); addCommandSupplier = new Supplier<Command<?>>() { @Override public Command<?> get() { return new CountingCommand(counter, 100); } }; int tasks = 100; for(int i = 0; i < tasks; i++) { defaultKeeperStateController.addKeeper(new KeeperTransMeta(getClusterId(), randomString(), new KeeperMeta())); } sleep(250); Assert.assertEquals(counter.get(), tasks); }