ShardingService { public void shardingIfNecessary() { if (!isNeedSharding()) { return; } if (!leaderElectionService.isLeader()) { blockUntilShardingCompleted(); return; } if (configService.isMonitorExecution()) { waitingOtherJobCompleted(); } log.debug("Elastic job: sharding begin."); jobNodeStorage.fillEphemeralJobNode(ShardingNode.PROCESSING, ""); clearShardingInfo(); JobShardingStrategy jobShardingStrategy = JobShardingStrategyFactory.getStrategy(globalConfigService, configService.getJobShardingStrategyClass()); JobShardingStrategyOption option = new JobShardingStrategyOption(jobName, configService.getShardingTotalCount(), configService.getShardingItemParameters()); jobNodeStorage.executeInTransaction(new PersistShardingInfoTransactionExecutionCallback(jobShardingStrategy.sharding(serverService.getAvailableServers(), option))); log.debug("Elastic job: sharding completed."); } ShardingService(final CoordinatorRegistryCenter coordinatorRegistryCenter, final JobConfiguration jobConfiguration); void setReshardingFlag(); boolean isNeedSharding(); void shardingIfNecessary(); List<Integer> getLocalHostShardingItems();  }
@Test public void assertShardingWhenUnnecessary() { when(jobNodeStorage.isJobNodeExisted("leader/sharding/necessary")).thenReturn(false); shardingService.shardingIfNecessary(); verify(jobNodeStorage).isJobNodeExisted("leader/sharding/necessary"); }
@Test public void assertShardingWhenIsNotLeaderAndIsShardingProcessing() { when(jobNodeStorage.isJobNodeExisted("leader/sharding/necessary")).thenReturn(true, true, false, false); when(leaderElectionService.isLeader()).thenReturn(false); when(jobNodeStorage.isJobNodeExisted("leader/sharding/processing")).thenReturn(true, false); shardingService.shardingIfNecessary(); verify(jobNodeStorage, times(4)).isJobNodeExisted("leader/sharding/necessary"); verify(jobNodeStorage, times(2)).isJobNodeExisted("leader/sharding/processing"); }
@Test public void assertShardingNecessaryWhenMonitorExecutionEnabled() { when(jobNodeStorage.isJobNodeExisted("leader/sharding/necessary")).thenReturn(true); when(leaderElectionService.isLeader()).thenReturn(true); when(configService.isMonitorExecution()).thenReturn(true); when(serverService.getAllServers()).thenReturn(Arrays.asList("ip1", "ip2")); when(executionService.hasRunningItems()).thenReturn(true, false); when(configService.getJobShardingStrategyClass()).thenReturn(AverageAllocationJobShardingStrategy.class.getCanonicalName()); when(configService.getShardingTotalCount()).thenReturn(3); when(configService.getShardingItemParameters()).thenReturn(Collections.<Integer, String>emptyMap()); shardingService.shardingIfNecessary(); verify(jobNodeStorage).isJobNodeExisted("leader/sharding/necessary"); verify(leaderElectionService).isLeader(); verify(configService).isMonitorExecution(); verify(executionService, times(2)).hasRunningItems(); verify(jobNodeStorage).removeJobNodeIfExisted("servers/ip1/sharding"); verify(jobNodeStorage).removeJobNodeIfExisted("servers/ip2/sharding"); verify(jobNodeStorage).fillEphemeralJobNode("leader/sharding/processing", ""); verify(configService).getJobShardingStrategyClass(); verify(configService).getShardingTotalCount(); verify(configService).getShardingItemParameters(); verify(jobNodeStorage).executeInTransaction(any(TransactionExecutionCallback.class)); }
@Test public void assertShardingNecessaryWhenMonitorExecutionDisabled() throws Exception { when(jobNodeStorage.isJobNodeExisted("leader/sharding/necessary")).thenReturn(true); when(leaderElectionService.isLeader()).thenReturn(true); when(configService.isMonitorExecution()).thenReturn(false); when(serverService.getAllServers()).thenReturn(Arrays.asList("ip1", "ip2")); when(configService.getJobShardingStrategyClass()).thenReturn(AverageAllocationJobShardingStrategy.class.getCanonicalName()); when(configService.getShardingTotalCount()).thenReturn(3); when(configService.getShardingItemParameters()).thenReturn(Collections.<Integer, String>emptyMap()); shardingService.shardingIfNecessary(); verify(jobNodeStorage).isJobNodeExisted("leader/sharding/necessary"); verify(leaderElectionService).isLeader(); verify(configService).isMonitorExecution(); verify(jobNodeStorage).removeJobNodeIfExisted("servers/ip1/sharding"); verify(jobNodeStorage).removeJobNodeIfExisted("servers/ip2/sharding"); verify(jobNodeStorage).fillEphemeralJobNode("leader/sharding/processing", ""); verify(configService).getJobShardingStrategyClass(); verify(configService).getShardingTotalCount(); verify(configService).getShardingItemParameters(); verify(jobNodeStorage).executeInTransaction(any(TransactionExecutionCallback.class)); }