@Test public void assertShardingNecessaryWhenMonitorExecutionEnabledAndIncreaseShardingTotalCount() { when(instanceService.getAvailableJobInstances()).thenReturn(Collections.singletonList(new JobInstance("127.0.0.1@-@0"))); when(jobNodeStorage.isJobNodeExisted("leader/sharding/necessary")).thenReturn(true); when(leaderService.isLeaderUntilBlock()).thenReturn(true); when(configService.load(false)).thenReturn(JobConfiguration.newBuilder("test_job", 3).cron("0/1 * * * * ?").monitorExecution(true).build()); when(executionService.hasRunningItems()).thenReturn(true, false); when(jobNodeStorage.getJobNodeChildrenKeys(ShardingNode.ROOT)).thenReturn(Arrays.asList("0", "1")); shardingService.shardingIfNecessary(); verify(executionService, times(2)).hasRunningItems(); verify(jobNodeStorage).removeJobNodeIfExisted("sharding/0/instance"); verify(jobNodeStorage).createJobNodeIfNeeded("sharding/0"); verify(jobNodeStorage).removeJobNodeIfExisted("sharding/1/instance"); verify(jobNodeStorage).createJobNodeIfNeeded("sharding/1"); verify(jobNodeStorage).removeJobNodeIfExisted("sharding/2/instance"); verify(jobNodeStorage).createJobNodeIfNeeded("sharding/2"); verify(jobNodeStorage).fillEphemeralJobNode("leader/sharding/processing", ""); verify(jobNodeStorage).executeInTransaction(any(TransactionExecutionCallback.class)); }
public void shardingIfNecessary() { List<JobInstance> availableJobInstances = instanceService.getAvailableJobInstances(); if (!isNeedSharding() || availableJobInstances.isEmpty()) { return; } if (!leaderService.isLeaderUntilBlock()) { blockUntilShardingCompleted(); return; } waitingOtherShardingItemCompleted(); JobConfiguration jobConfig = configService.load(false); int shardingTotalCount = jobConfig.getShardingTotalCount(); log.debug("Job '{}' sharding begin.", jobName); jobNodeStorage.fillEphemeralJobNode(ShardingNode.PROCESSING, ""); resetShardingInfo(shardingTotalCount); JobShardingStrategy jobShardingStrategy = JobShardingStrategyFactory.getStrategy(jobConfig.getJobShardingStrategyType()); jobNodeStorage.executeInTransaction(new PersistShardingInfoTransactionExecutionCallback(jobShardingStrategy.sharding(availableJobInstances, jobName, shardingTotalCount))); log.debug("Job '{}' sharding complete.", jobName); }
ShardingService { public void shardingIfNecessary() { List<JobInstance> availableJobInstances = instanceService.getAvailableJobInstances(); if (!isNeedSharding() || availableJobInstances.isEmpty()) { return; } if (!leaderService.isLeaderUntilBlock()) { blockUntilShardingCompleted(); return; } waitingOtherShardingItemCompleted(); JobConfiguration jobConfig = configService.load(false); int shardingTotalCount = jobConfig.getShardingTotalCount(); log.debug("Job '{}' sharding begin.", jobName); jobNodeStorage.fillEphemeralJobNode(ShardingNode.PROCESSING, ""); resetShardingInfo(shardingTotalCount); JobShardingStrategy jobShardingStrategy = JobShardingStrategyFactory.getStrategy(jobConfig.getJobShardingStrategyType()); jobNodeStorage.executeInTransaction(new PersistShardingInfoTransactionExecutionCallback(jobShardingStrategy.sharding(availableJobInstances, jobName, shardingTotalCount))); log.debug("Job '{}' sharding complete.", jobName); } }
ShardingService { public void shardingIfNecessary() { List<JobInstance> availableJobInstances = instanceService.getAvailableJobInstances(); if (!isNeedSharding() || availableJobInstances.isEmpty()) { return; } if (!leaderService.isLeaderUntilBlock()) { blockUntilShardingCompleted(); return; } waitingOtherShardingItemCompleted(); JobConfiguration jobConfig = configService.load(false); int shardingTotalCount = jobConfig.getShardingTotalCount(); log.debug("Job '{}' sharding begin.", jobName); jobNodeStorage.fillEphemeralJobNode(ShardingNode.PROCESSING, ""); resetShardingInfo(shardingTotalCount); JobShardingStrategy jobShardingStrategy = JobShardingStrategyFactory.getStrategy(jobConfig.getJobShardingStrategyType()); jobNodeStorage.executeInTransaction(new PersistShardingInfoTransactionExecutionCallback(jobShardingStrategy.sharding(availableJobInstances, jobName, shardingTotalCount))); log.debug("Job '{}' sharding complete.", jobName); } ShardingService(final CoordinatorRegistryCenter regCenter, final String jobName); }
ShardingService { public void shardingIfNecessary() { List<JobInstance> availableJobInstances = instanceService.getAvailableJobInstances(); if (!isNeedSharding() || availableJobInstances.isEmpty()) { return; } if (!leaderService.isLeaderUntilBlock()) { blockUntilShardingCompleted(); return; } waitingOtherShardingItemCompleted(); JobConfiguration jobConfig = configService.load(false); int shardingTotalCount = jobConfig.getShardingTotalCount(); log.debug("Job '{}' sharding begin.", jobName); jobNodeStorage.fillEphemeralJobNode(ShardingNode.PROCESSING, ""); resetShardingInfo(shardingTotalCount); JobShardingStrategy jobShardingStrategy = JobShardingStrategyFactory.getStrategy(jobConfig.getJobShardingStrategyType()); jobNodeStorage.executeInTransaction(new PersistShardingInfoTransactionExecutionCallback(jobShardingStrategy.sharding(availableJobInstances, jobName, shardingTotalCount))); log.debug("Job '{}' sharding complete.", jobName); } ShardingService(final CoordinatorRegistryCenter regCenter, final String jobName); void setReshardingFlag(); boolean isNeedSharding(); void shardingIfNecessary(); List<Integer> getShardingItems(final String jobInstanceId); List<Integer> getLocalShardingItems(); boolean hasShardingInfoInOfflineServers(); }
ShardingService { public void shardingIfNecessary() { List<JobInstance> availableJobInstances = instanceService.getAvailableJobInstances(); if (!isNeedSharding() || availableJobInstances.isEmpty()) { return; } if (!leaderService.isLeaderUntilBlock()) { blockUntilShardingCompleted(); return; } waitingOtherShardingItemCompleted(); JobConfiguration jobConfig = configService.load(false); int shardingTotalCount = jobConfig.getShardingTotalCount(); log.debug("Job '{}' sharding begin.", jobName); jobNodeStorage.fillEphemeralJobNode(ShardingNode.PROCESSING, ""); resetShardingInfo(shardingTotalCount); JobShardingStrategy jobShardingStrategy = JobShardingStrategyFactory.getStrategy(jobConfig.getJobShardingStrategyType()); jobNodeStorage.executeInTransaction(new PersistShardingInfoTransactionExecutionCallback(jobShardingStrategy.sharding(availableJobInstances, jobName, shardingTotalCount))); log.debug("Job '{}' sharding complete.", jobName); } ShardingService(final CoordinatorRegistryCenter regCenter, final String jobName); void setReshardingFlag(); boolean isNeedSharding(); void shardingIfNecessary(); List<Integer> getShardingItems(final String jobInstanceId); List<Integer> getLocalShardingItems(); boolean hasShardingInfoInOfflineServers();  }