@Test public void testStopObserving() throws Exception { client.startObserving("shard2", "shard1", 1000); client.stopObserving("shard2", "shard1", 1000); List<Integer> memberIds = config.getMembersInShard("shard2").stream().map(Config.ConfigMember::getMemberId) .collect(Collectors.toList()); for (Config.ConfigMember m : config.getMembersInShard("shard2")) { ShardManager shardManager = shardManagers.get(m.getHostId()); verify(shardManager).stopObserving(any(), any(), anyLong()); } for (ChildData d : stats.getCurrentData()) { ZookeeperStat stat = objectMapper.readValue(d.getData(), ZookeeperStat.class); if (!memberIds.contains(stat.memberId)) { continue; } assertTrue(stat.isNormalOperational()); } }
@Override public void stopObserving(String shardId, String observedShardId, long timeoutMs) throws ShardManagerException, InterruptedException { sendActionToShard(shardId, STOP_SLAVE, shardId, observedShardId, timeoutMs); waitCondition(shardId, ZookeeperStat::isNormalOperational, timeoutMs); }
ZookeeperShardManagerClient implements ShardManagerClient { @Override public void stopObserving(String shardId, String observedShardId, long timeoutMs) throws ShardManagerException, InterruptedException { sendActionToShard(shardId, STOP_SLAVE, shardId, observedShardId, timeoutMs); waitCondition(shardId, ZookeeperStat::isNormalOperational, timeoutMs); } }
ZookeeperShardManagerClient implements ShardManagerClient { @Override public void stopObserving(String shardId, String observedShardId, long timeoutMs) throws ShardManagerException, InterruptedException { sendActionToShard(shardId, STOP_SLAVE, shardId, observedShardId, timeoutMs); waitCondition(shardId, ZookeeperStat::isNormalOperational, timeoutMs); } ZookeeperShardManagerClient(String serviceName, String clientName, String connectString, Config config); }
ZookeeperShardManagerClient implements ShardManagerClient { @Override public void stopObserving(String shardId, String observedShardId, long timeoutMs) throws ShardManagerException, InterruptedException { sendActionToShard(shardId, STOP_SLAVE, shardId, observedShardId, timeoutMs); waitCondition(shardId, ZookeeperStat::isNormalOperational, timeoutMs); } 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 stopObserving(String shardId, String observedShardId, long timeoutMs) throws ShardManagerException, InterruptedException { sendActionToShard(shardId, STOP_SLAVE, shardId, observedShardId, timeoutMs); waitCondition(shardId, ZookeeperStat::isNormalOperational, timeoutMs); } 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();  }