ShardManager implements ShardManagerProtocol { @Override public void stopObserving(String shardId, String masterShardId, long timeoutMs) throws ShardManagerException, InterruptedException { trace("[{}] shardId={} un-followed shardId={}", gondola.getHostId(), shardId, masterShardId); Member.SlaveStatus status = gondola.getShard(shardId).getLocalMember().getSlaveStatus(); if (status == null) { return; } String curMasterShardId = config.getMember(status.masterId).getShardId(); if (!curMasterShardId.equals(masterShardId)) { throw new ShardManagerException(FAILED_STOP_SLAVE, String.format( "Cannot stop slave due to follow different shard. current=%s, target=%s", curMasterShardId, masterShardId)); } try { gondola.getShard(shardId).getLocalMember().setSlave(-1); } catch (GondolaException e) { throw new ShardManagerException(e); } try { if (!Utils.pollingWithTimeout(() -> { if (gondola.getShard(shardId).getLocalMember().getSlaveStatus() == null) { return true; } logger.warn("Failed stop observing {} on shard={}", masterShardId, shardId); return false; }, timeoutMs / POLLING_TIMES, timeoutMs)) { throw new ShardManagerException(FAILED_STOP_SLAVE, "timed out"); } } catch (ExecutionException e) { throw new ShardManagerException(e); } observedShards.remove(masterShardId); } 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; }
@Test public void testStopObserving() throws Exception { Member.SlaveStatus status = getSuccessSlaveStatus(); when(member.getSlaveStatus()).thenReturn(status); shardManager.startObserving(FROM_SHARD, TARGET_SHARD, 300); assertTrue(getObservedShards().contains(TARGET_SHARD)); when(member.getSlaveStatus()).thenReturn(status).thenReturn(null); shardManager.stopObserving(FROM_SHARD, TARGET_SHARD, 300); assertFalse(getObservedShards().contains(TARGET_SHARD)); when(member.getSlaveStatus()).thenReturn(null); shardManager.stopObserving(FROM_SHARD, TARGET_SHARD, 300); assertFalse(getObservedShards().contains(TARGET_SHARD)); }