@Test public void testStartObserving_success() throws Exception { Member.SlaveStatus status = getSuccessSlaveStatus(); when(member.getSlaveStatus()).thenReturn(status); assertFalse(getObservedShards().contains(TARGET_SHARD)); shardManager.startObserving(FROM_SHARD, TARGET_SHARD, 300); assertTrue(getObservedShards().contains(TARGET_SHARD)); shardManager.startObserving(FROM_SHARD, TARGET_SHARD, 300); assertTrue(getObservedShards().contains(TARGET_SHARD)); }
@Override public void startObserving(String shardId, String observedShardId, long timeoutMs) throws ShardManagerException, InterruptedException { boolean success = false; trace("[{}-{}] Try to follow shardId={} as slave...", gondola.getHostId(), gondola.getShard(shardId).getLocalMember().getMemberId(), observedShardId); List<Config.ConfigMember> membersInShard = config.getMembersInShard(observedShardId); for (Config.ConfigMember m : membersInShard) { if (success = setSlave(shardId, m.getMemberId(), timeoutMs / membersInShard.size())) { filter.getChangeLogProcessor().reset(shardId); trace("[{}-{}] Successfully to follow masterId={}", gondola.getHostId(), gondola.getShard(shardId).getLocalMember().getMemberId(), m.getMemberId()); break; } } if (!success) { logger.error("[{}-{}] Failed follow master={}", gondola.getHostId(), gondola.getShard(shardId).getLocalMember().getMemberId(), observedShardId); throw new ShardManagerException(FAILED_START_SLAVE); } observedShards.add(observedShardId); }
ShardManager implements ShardManagerProtocol { @Override public void startObserving(String shardId, String observedShardId, long timeoutMs) throws ShardManagerException, InterruptedException { boolean success = false; trace("[{}-{}] Try to follow shardId={} as slave...", gondola.getHostId(), gondola.getShard(shardId).getLocalMember().getMemberId(), observedShardId); List<Config.ConfigMember> membersInShard = config.getMembersInShard(observedShardId); for (Config.ConfigMember m : membersInShard) { if (success = setSlave(shardId, m.getMemberId(), timeoutMs / membersInShard.size())) { filter.getChangeLogProcessor().reset(shardId); trace("[{}-{}] Successfully to follow masterId={}", gondola.getHostId(), gondola.getShard(shardId).getLocalMember().getMemberId(), m.getMemberId()); break; } } if (!success) { logger.error("[{}-{}] Failed follow master={}", gondola.getHostId(), gondola.getShard(shardId).getLocalMember().getMemberId(), observedShardId); throw new ShardManagerException(FAILED_START_SLAVE); } observedShards.add(observedShardId); } }
ShardManager implements ShardManagerProtocol { @Override public void startObserving(String shardId, String observedShardId, long timeoutMs) throws ShardManagerException, InterruptedException { boolean success = false; trace("[{}-{}] Try to follow shardId={} as slave...", gondola.getHostId(), gondola.getShard(shardId).getLocalMember().getMemberId(), observedShardId); List<Config.ConfigMember> membersInShard = config.getMembersInShard(observedShardId); for (Config.ConfigMember m : membersInShard) { if (success = setSlave(shardId, m.getMemberId(), timeoutMs / membersInShard.size())) { filter.getChangeLogProcessor().reset(shardId); trace("[{}-{}] Successfully to follow masterId={}", gondola.getHostId(), gondola.getShard(shardId).getLocalMember().getMemberId(), m.getMemberId()); break; } } if (!success) { logger.error("[{}-{}] Failed follow master={}", gondola.getHostId(), gondola.getShard(shardId).getLocalMember().getMemberId(), observedShardId); throw new ShardManagerException(FAILED_START_SLAVE); } observedShards.add(observedShardId); } ShardManager(Gondola gondola, RoutingFilter filter, Config config, ShardManagerClient shardManagerClient); }
ShardManager implements ShardManagerProtocol { @Override public void startObserving(String shardId, String observedShardId, long timeoutMs) throws ShardManagerException, InterruptedException { boolean success = false; trace("[{}-{}] Try to follow shardId={} as slave...", gondola.getHostId(), gondola.getShard(shardId).getLocalMember().getMemberId(), observedShardId); List<Config.ConfigMember> membersInShard = config.getMembersInShard(observedShardId); for (Config.ConfigMember m : membersInShard) { if (success = setSlave(shardId, m.getMemberId(), timeoutMs / membersInShard.size())) { filter.getChangeLogProcessor().reset(shardId); trace("[{}-{}] Successfully to follow masterId={}", gondola.getHostId(), gondola.getShard(shardId).getLocalMember().getMemberId(), m.getMemberId()); break; } } if (!success) { logger.error("[{}-{}] Failed follow master={}", gondola.getHostId(), gondola.getShard(shardId).getLocalMember().getMemberId(), observedShardId); throw new ShardManagerException(FAILED_START_SLAVE); } observedShards.add(observedShardId); } ShardManager(Gondola gondola, RoutingFilter filter, Config config, ShardManagerClient shardManagerClient); @Override void startObserving(String shardId, String observedShardId, long timeoutMs); @Override void stopObserving(String shardId, String masterShardId, 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); ShardManagerClient getShardManagerClient(); }
ShardManager implements ShardManagerProtocol { @Override public void startObserving(String shardId, String observedShardId, long timeoutMs) throws ShardManagerException, InterruptedException { boolean success = false; trace("[{}-{}] Try to follow shardId={} as slave...", gondola.getHostId(), gondola.getShard(shardId).getLocalMember().getMemberId(), observedShardId); List<Config.ConfigMember> membersInShard = config.getMembersInShard(observedShardId); for (Config.ConfigMember m : membersInShard) { if (success = setSlave(shardId, m.getMemberId(), timeoutMs / membersInShard.size())) { filter.getChangeLogProcessor().reset(shardId); trace("[{}-{}] Successfully to follow masterId={}", gondola.getHostId(), gondola.getShard(shardId).getLocalMember().getMemberId(), m.getMemberId()); break; } } if (!success) { logger.error("[{}-{}] Failed follow master={}", gondola.getHostId(), gondola.getShard(shardId).getLocalMember().getMemberId(), observedShardId); throw new ShardManagerException(FAILED_START_SLAVE); } observedShards.add(observedShardId); } ShardManager(Gondola gondola, RoutingFilter filter, Config config, ShardManagerClient shardManagerClient); @Override void startObserving(String shardId, String observedShardId, long timeoutMs); @Override void stopObserving(String shardId, String masterShardId, 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); ShardManagerClient getShardManagerClient(); static final int POLLING_TIMES; static final int LOG_APPROACHING_DIFF; }