@Test public void testMigrateBuckets() throws Exception { client.migrateBuckets(Range.closed(0, 10), "shard1", "shard2", 1000); for (Config.ConfigMember m : config.getMembersInShard("shard1")) { ShardManager shardManager = shardManagers.get(m.getHostId()); verify(shardManager).migrateBuckets(any(), any(), any(), anyLong()); } }
@Override public void migrateBuckets(Range<Integer> splitRange, String fromShardId, String toShardId, long timeoutMs) throws ShardManagerException, InterruptedException { sendActionToShard(fromShardId, MIGRATE_1, splitRange.lowerEndpoint(), splitRange.upperEndpoint(), fromShardId, toShardId, timeoutMs); if (!waitCondition(fromShardId, ZookeeperStat::isMigrating1Operational, timeoutMs)) { throw new ShardManagerException(FAILED_MIGRATE_1, "Timed out"); } setBuckets(splitRange, fromShardId, toShardId, false); setBuckets(splitRange, fromShardId, toShardId, true); sendActionToAll(Action.NOOP); }
ZookeeperShardManagerClient implements ShardManagerClient { @Override public void migrateBuckets(Range<Integer> splitRange, String fromShardId, String toShardId, long timeoutMs) throws ShardManagerException, InterruptedException { sendActionToShard(fromShardId, MIGRATE_1, splitRange.lowerEndpoint(), splitRange.upperEndpoint(), fromShardId, toShardId, timeoutMs); if (!waitCondition(fromShardId, ZookeeperStat::isMigrating1Operational, timeoutMs)) { throw new ShardManagerException(FAILED_MIGRATE_1, "Timed out"); } setBuckets(splitRange, fromShardId, toShardId, false); setBuckets(splitRange, fromShardId, toShardId, true); sendActionToAll(Action.NOOP); } }
ZookeeperShardManagerClient implements ShardManagerClient { @Override public void migrateBuckets(Range<Integer> splitRange, String fromShardId, String toShardId, long timeoutMs) throws ShardManagerException, InterruptedException { sendActionToShard(fromShardId, MIGRATE_1, splitRange.lowerEndpoint(), splitRange.upperEndpoint(), fromShardId, toShardId, timeoutMs); if (!waitCondition(fromShardId, ZookeeperStat::isMigrating1Operational, timeoutMs)) { throw new ShardManagerException(FAILED_MIGRATE_1, "Timed out"); } setBuckets(splitRange, fromShardId, toShardId, false); setBuckets(splitRange, fromShardId, toShardId, true); sendActionToAll(Action.NOOP); } ZookeeperShardManagerClient(String serviceName, String clientName, String connectString, Config config); }
ZookeeperShardManagerClient implements ShardManagerClient { @Override public void migrateBuckets(Range<Integer> splitRange, String fromShardId, String toShardId, long timeoutMs) throws ShardManagerException, InterruptedException { sendActionToShard(fromShardId, MIGRATE_1, splitRange.lowerEndpoint(), splitRange.upperEndpoint(), fromShardId, toShardId, timeoutMs); if (!waitCondition(fromShardId, ZookeeperStat::isMigrating1Operational, timeoutMs)) { throw new ShardManagerException(FAILED_MIGRATE_1, "Timed out"); } setBuckets(splitRange, fromShardId, toShardId, false); setBuckets(splitRange, fromShardId, toShardId, true); sendActionToAll(Action.NOOP); } 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 migrateBuckets(Range<Integer> splitRange, String fromShardId, String toShardId, long timeoutMs) throws ShardManagerException, InterruptedException { sendActionToShard(fromShardId, MIGRATE_1, splitRange.lowerEndpoint(), splitRange.upperEndpoint(), fromShardId, toShardId, timeoutMs); if (!waitCondition(fromShardId, ZookeeperStat::isMigrating1Operational, timeoutMs)) { throw new ShardManagerException(FAILED_MIGRATE_1, "Timed out"); } setBuckets(splitRange, fromShardId, toShardId, false); setBuckets(splitRange, fromShardId, toShardId, true); sendActionToAll(Action.NOOP); } 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();  }