@Test public void testSetBuckets() throws Exception { client.setBuckets(Range.closed(0, 10), "shard1", "shard2", false); for (String hostId : config.getHostIds()) { verify(shardManagers.get(hostId)).setBuckets(any(), any(),any(), anyBoolean()); } }
@Override public void setBuckets(Range<Integer> splitRange, String fromShardId, String toShardId, boolean migrationComplete) throws ShardManagerException, InterruptedException { sendActionToAll(!migrationComplete ? MIGRATE_2 : MIGRATE_3, splitRange.lowerEndpoint(), splitRange.upperEndpoint(), fromShardId, toShardId, migrationComplete); if (!waitCondition(null, !migrationComplete ? ZookeeperStat::isMigrating2Operational : ZookeeperStat::isNormalOperational, 3000)) { throw new ShardManagerException(FAILED_SET_BUCKETS, "timed out"); } }
ZookeeperShardManagerClient implements ShardManagerClient { @Override public void setBuckets(Range<Integer> splitRange, String fromShardId, String toShardId, boolean migrationComplete) throws ShardManagerException, InterruptedException { sendActionToAll(!migrationComplete ? MIGRATE_2 : MIGRATE_3, splitRange.lowerEndpoint(), splitRange.upperEndpoint(), fromShardId, toShardId, migrationComplete); if (!waitCondition(null, !migrationComplete ? ZookeeperStat::isMigrating2Operational : ZookeeperStat::isNormalOperational, 3000)) { throw new ShardManagerException(FAILED_SET_BUCKETS, "timed out"); } } }
ZookeeperShardManagerClient implements ShardManagerClient { @Override public void setBuckets(Range<Integer> splitRange, String fromShardId, String toShardId, boolean migrationComplete) throws ShardManagerException, InterruptedException { sendActionToAll(!migrationComplete ? MIGRATE_2 : MIGRATE_3, splitRange.lowerEndpoint(), splitRange.upperEndpoint(), fromShardId, toShardId, migrationComplete); if (!waitCondition(null, !migrationComplete ? ZookeeperStat::isMigrating2Operational : ZookeeperStat::isNormalOperational, 3000)) { throw new ShardManagerException(FAILED_SET_BUCKETS, "timed out"); } } ZookeeperShardManagerClient(String serviceName, String clientName, String connectString, Config config); }
ZookeeperShardManagerClient implements ShardManagerClient { @Override public void setBuckets(Range<Integer> splitRange, String fromShardId, String toShardId, boolean migrationComplete) throws ShardManagerException, InterruptedException { sendActionToAll(!migrationComplete ? MIGRATE_2 : MIGRATE_3, splitRange.lowerEndpoint(), splitRange.upperEndpoint(), fromShardId, toShardId, migrationComplete); if (!waitCondition(null, !migrationComplete ? ZookeeperStat::isMigrating2Operational : ZookeeperStat::isNormalOperational, 3000)) { throw new ShardManagerException(FAILED_SET_BUCKETS, "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 setBuckets(Range<Integer> splitRange, String fromShardId, String toShardId, boolean migrationComplete) throws ShardManagerException, InterruptedException { sendActionToAll(!migrationComplete ? MIGRATE_2 : MIGRATE_3, splitRange.lowerEndpoint(), splitRange.upperEndpoint(), fromShardId, toShardId, migrationComplete); if (!waitCondition(null, !migrationComplete ? ZookeeperStat::isMigrating2Operational : ZookeeperStat::isNormalOperational, 3000)) { throw new ShardManagerException(FAILED_SET_BUCKETS, "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();  }