@Test public void testStartObserving() throws Exception { client.startObserving("shard2", "shard1", 1000); List<Integer> memberIds = config.getMembersInShard("shard2").stream().map(Config.ConfigMember::getMemberId) .collect(Collectors.toList()); for (Integer memberId : memberIds) { ShardManager shardManager = shardManagers.get(config.getMember(memberId).getHostId()); verify(shardManager).startObserving(any(), any(), anyLong()); } for (ChildData d : stats.getCurrentData()) { ZookeeperStat stat = objectMapper.readValue(d.getData(), ZookeeperStat.class); if (!memberIds.contains(stat.memberId)) { continue; } assertTrue(stat.isSlaveOperational()); } }
@Override public void startObserving(String shardId, String observedShardId, long timeoutMs) throws ShardManagerException, InterruptedException { sendActionToShard(shardId, START_SLAVE, shardId, observedShardId, timeoutMs); if (!waitCondition(shardId, ZookeeperStat::isSlaveOperational, timeoutMs)) { throw new ShardManagerException(FAILED_START_SLAVE, "timed out"); } }
ZookeeperShardManagerClient implements ShardManagerClient { @Override public void startObserving(String shardId, String observedShardId, long timeoutMs) throws ShardManagerException, InterruptedException { sendActionToShard(shardId, START_SLAVE, shardId, observedShardId, timeoutMs); if (!waitCondition(shardId, ZookeeperStat::isSlaveOperational, timeoutMs)) { throw new ShardManagerException(FAILED_START_SLAVE, "timed out"); } } }
ZookeeperShardManagerClient implements ShardManagerClient { @Override public void startObserving(String shardId, String observedShardId, long timeoutMs) throws ShardManagerException, InterruptedException { sendActionToShard(shardId, START_SLAVE, shardId, observedShardId, timeoutMs); if (!waitCondition(shardId, ZookeeperStat::isSlaveOperational, timeoutMs)) { throw new ShardManagerException(FAILED_START_SLAVE, "timed out"); } } ZookeeperShardManagerClient(String serviceName, String clientName, String connectString, Config config); }
ZookeeperShardManagerClient implements ShardManagerClient { @Override public void startObserving(String shardId, String observedShardId, long timeoutMs) throws ShardManagerException, InterruptedException { sendActionToShard(shardId, START_SLAVE, shardId, observedShardId, timeoutMs); if (!waitCondition(shardId, ZookeeperStat::isSlaveOperational, timeoutMs)) { throw new ShardManagerException(FAILED_START_SLAVE, "timed out"); } } ZookeeperShardManagerClient(String serviceName, String clientName, String connectString, Config config); @Override void startObserving(String shardId, String observedShardId, long timeoutMs); @Override void stopObserving(String shardId, String observedShardId, long timeoutMs); @Override void migrateBuckets(Range<Integer> splitRange, String fromShardId, String toShardId, long timeoutMs); @Override boolean waitSlavesSynced(String shardId, long timeoutMs); @Override boolean waitSlavesApproaching(String shardId, long timeoutMs); @Override void setBuckets(Range<Integer> splitRange, String fromShardId, String toShardId, boolean migrationComplete); @Override void rollbackBuckets(Range<Integer> splitRange); @Override void stop(); }
ZookeeperShardManagerClient implements ShardManagerClient { @Override public void startObserving(String shardId, String observedShardId, long timeoutMs) throws ShardManagerException, InterruptedException { sendActionToShard(shardId, START_SLAVE, shardId, observedShardId, timeoutMs); if (!waitCondition(shardId, ZookeeperStat::isSlaveOperational, timeoutMs)) { throw new ShardManagerException(FAILED_START_SLAVE, "timed out"); } } ZookeeperShardManagerClient(String serviceName, String clientName, String connectString, Config config); @Override void startObserving(String shardId, String observedShardId, long timeoutMs); @Override void stopObserving(String shardId, String observedShardId, long timeoutMs); @Override void migrateBuckets(Range<Integer> splitRange, String fromShardId, String toShardId, long timeoutMs); @Override boolean waitSlavesSynced(String shardId, long timeoutMs); @Override boolean waitSlavesApproaching(String shardId, long timeoutMs); @Override void setBuckets(Range<Integer> splitRange, String fromShardId, String toShardId, boolean migrationComplete); @Override void rollbackBuckets(Range<Integer> splitRange); @Override void stop();  }