@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"); }
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 { 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 { 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); }
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(); }
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();  }