DefaultChangePrimaryDcAction implements ChangePrimaryDcAction { @Override public PrimaryDcChangeMessage changePrimaryDc(String clusterId, String shardId, String newPrimaryDc, MasterInfo masterInfo) { if(!currentMetaManager.hasCluster(clusterId)){ logger.info("[changePrimaryDc][not interested in this cluster]"); return new PrimaryDcChangeMessage(PRIMARY_DC_CHANGE_RESULT.SUCCESS, "not interested in this cluster:" + clusterId); } ChangePrimaryDcAction changePrimaryDcAction = null; ExecutionLog executionLog = new ExecutionLog(String.format("meta server:%s", currentClusterServer.getClusterInfo())); if(newPrimaryDc.equalsIgnoreCase(dcMetaCache.getCurrentDc())){ logger.info("[doChangePrimaryDc][become primary]{}, {}, {}", clusterId, shardId, newPrimaryDc); changePrimaryDcAction = new BecomePrimaryAction(dcMetaCache, currentMetaManager, sentinelManager, offsetWaiter, executionLog, keyedObjectPool, createNewMasterChooser(), scheduled, executors); ChangePrimaryDcJob changePrimaryDcJob = createChangePrimaryDcJob(changePrimaryDcAction, clusterId, shardId, newPrimaryDc, masterInfo); int timeout = DEFAULT_SO_TIMEOUT / 2; try { clusterShardExecutors.clearAndExecute(new Pair<>(clusterId, shardId), changePrimaryDcJob); waitForCommandStart(changePrimaryDcJob); return changePrimaryDcJob.future().get(timeout, TimeUnit.MILLISECONDS); } catch (TimeoutException|InterruptedException e) { logger.error("[changePrimaryDc][execute may timeout][fall to run directly]{}, {}, {}", clusterId, shardId, newPrimaryDc, e); try { return createChangePrimaryDcJob(changePrimaryDcAction, clusterId, shardId, newPrimaryDc, masterInfo) .execute().get(); } catch (Exception innerException) { logger.error("[changePrimaryDc][try direct-run failed]{}, {}, {}", clusterId, shardId, newPrimaryDc, e); return new PrimaryDcChangeMessage(PRIMARY_DC_CHANGE_RESULT.FAIL, executionLog.getLog()); } }catch (Exception e) { logger.error("[changePrimaryDc][execute by adjust executors fail]" + clusterId + "," + shardId + "," + newPrimaryDc, e); return new PrimaryDcChangeMessage(PRIMARY_DC_CHANGE_RESULT.FAIL, executionLog.getLog()); } } else { logger.info("[doChangePrimaryDc][become backup]{}, {}, {}", clusterId, shardId, newPrimaryDc); changePrimaryDcAction = new BecomeBackupAction(dcMetaCache, currentMetaManager, sentinelManager, executionLog, keyedObjectPool, multiDcService, scheduled, executors); return changePrimaryDcAction.changePrimaryDc(clusterId, shardId, newPrimaryDc, masterInfo); } }  @Override PrimaryDcChangeMessage changePrimaryDc(String clusterId, String shardId, String newPrimaryDc, MasterInfo masterInfo);  }
@Test public void testChangePrimaryDcFirstTimeout() { when(currentMetaManager.hasCluster(anyString())).thenReturn(true); when(dcMetaCache.getCurrentDc()).thenReturn("SHAOY"); MetaServerConsoleService.PrimaryDcChangeMessage result = action.changePrimaryDc("cluster", "shard", "SHAOY", new MasterInfo()); Assert.assertEquals(expectedResult, result); }