ExecutionService { public void cleanPreviousExecutionInfo() { if (!jobNodeStorage.isJobNodeExisted(ExecutionNode.ROOT)) { return; } if (leaderElectionService.isLeader()) { jobNodeStorage.fillEphemeralJobNode(ExecutionNode.CLEANING, ""); List<Integer> items = getAllItems(); for (int each : items) { jobNodeStorage.removeJobNodeIfExisted(ExecutionNode.getCompletedNode(each)); } if (jobNodeStorage.isJobNodeExisted(ExecutionNode.NECESSARY)) { fixExecutionInfo(items); } jobNodeStorage.removeJobNodeIfExisted(ExecutionNode.CLEANING); } while (jobNodeStorage.isJobNodeExisted(ExecutionNode.CLEANING)) { BlockUtils.waitingShortTime(); } } ExecutionService(final CoordinatorRegistryCenter coordinatorRegistryCenter, final JobConfiguration jobConfiguration); void registerJobBegin(final JobExecutionMultipleShardingContext jobExecutionShardingContext); void cleanPreviousExecutionInfo(); void registerJobCompleted(final JobExecutionMultipleShardingContext jobExecutionShardingContext); void setNeedFixExecutionInfoFlag(); void clearRunningInfo(final List<Integer> items); boolean misfireIfNecessary(final List<Integer> items); void setMisfire(final List<Integer> items); List<Integer> getMisfiredJobItems(final List<Integer> items); void clearMisfire(final List<Integer> items); void removeExecutionInfo(); boolean isCompleted(final int item); boolean hasRunningItems(final List<Integer> items); boolean hasRunningItems();  }
@Test public void assertCleanPreviousExecutionInfoWhenNotMonitorExecution() { when(jobNodeStorage.isJobNodeExisted("execution")).thenReturn(false); executionService.cleanPreviousExecutionInfo(); verify(jobNodeStorage).isJobNodeExisted("execution"); verify(leaderElectionService, times(0)).isLeader(); }
@Test public void assertCleanPreviousExecutionInfoWhenIsNotLeader() { when(jobNodeStorage.isJobNodeExisted("execution")).thenReturn(true); when(leaderElectionService.isLeader()).thenReturn(false); when(jobNodeStorage.isJobNodeExisted("leader/execution/cleaning")).thenReturn(true, false); executionService.cleanPreviousExecutionInfo(); verify(jobNodeStorage).isJobNodeExisted("execution"); verify(leaderElectionService).isLeader(); verify(jobNodeStorage, times(2)).isJobNodeExisted("leader/execution/cleaning"); }
@Test public void assertCleanPreviousExecutionInfoWhenIsLeaderButNotNeedFixExecutionInfo() { when(jobNodeStorage.isJobNodeExisted("execution")).thenReturn(true); when(leaderElectionService.isLeader()).thenReturn(true); when(jobNodeStorage.getJobNodeChildrenKeys("execution")).thenReturn(Arrays.asList("0", "1", "2")); when(jobNodeStorage.isJobNodeExisted("leader/execution/necessary")).thenReturn(false); executionService.cleanPreviousExecutionInfo(); verify(jobNodeStorage).isJobNodeExisted("execution"); verify(leaderElectionService).isLeader(); verify(jobNodeStorage).fillEphemeralJobNode("leader/execution/cleaning", ""); verify(jobNodeStorage).getJobNodeChildrenKeys("execution"); verify(jobNodeStorage).removeJobNodeIfExisted("execution/0/completed"); verify(jobNodeStorage).removeJobNodeIfExisted("execution/1/completed"); verify(jobNodeStorage).removeJobNodeIfExisted("execution/2/completed"); verify(jobNodeStorage).isJobNodeExisted("leader/execution/necessary"); verify(jobNodeStorage).removeJobNodeIfExisted("leader/execution/cleaning"); verify(jobNodeStorage).isJobNodeExisted("leader/execution/cleaning"); }
@Test public void assertCleanPreviousExecutionInfoWhenNeedFixExecutionInfoForNewValuesGreater() { when(jobNodeStorage.isJobNodeExisted("execution")).thenReturn(true); when(leaderElectionService.isLeader()).thenReturn(true); when(jobNodeStorage.getJobNodeChildrenKeys("execution")).thenReturn(Arrays.asList("0", "1", "2")); when(jobNodeStorage.isJobNodeExisted("leader/execution/necessary")).thenReturn(true); when(configService.getShardingTotalCount()).thenReturn(4); executionService.cleanPreviousExecutionInfo(); verify(jobNodeStorage).isJobNodeExisted("execution"); verify(leaderElectionService).isLeader(); verify(jobNodeStorage).fillEphemeralJobNode("leader/execution/cleaning", ""); verify(jobNodeStorage).getJobNodeChildrenKeys("execution"); verify(jobNodeStorage).removeJobNodeIfExisted("execution/0/completed"); verify(jobNodeStorage).removeJobNodeIfExisted("execution/1/completed"); verify(jobNodeStorage).removeJobNodeIfExisted("execution/2/completed"); verify(jobNodeStorage).isJobNodeExisted("leader/execution/necessary"); verify(configService).getShardingTotalCount(); verify(jobNodeStorage).createJobNodeIfNeeded("execution/3"); verify(jobNodeStorage).removeJobNodeIfExisted("leader/execution/necessary"); verify(jobNodeStorage).removeJobNodeIfExisted("leader/execution/cleaning"); verify(jobNodeStorage).isJobNodeExisted("leader/execution/cleaning"); }
@Test public void assertCleanPreviousExecutionInfoWhenNeedFixExecutionInfoForNewValuesLess() { when(jobNodeStorage.isJobNodeExisted("execution")).thenReturn(true); when(leaderElectionService.isLeader()).thenReturn(true); when(jobNodeStorage.getJobNodeChildrenKeys("execution")).thenReturn(Arrays.asList("0", "1", "2")); when(jobNodeStorage.isJobNodeExisted("leader/execution/necessary")).thenReturn(true); when(configService.getShardingTotalCount()).thenReturn(2); executionService.cleanPreviousExecutionInfo(); verify(jobNodeStorage).isJobNodeExisted("execution"); verify(leaderElectionService).isLeader(); verify(jobNodeStorage).fillEphemeralJobNode("leader/execution/cleaning", ""); verify(jobNodeStorage).getJobNodeChildrenKeys("execution"); verify(jobNodeStorage).removeJobNodeIfExisted("execution/0/completed"); verify(jobNodeStorage).removeJobNodeIfExisted("execution/1/completed"); verify(jobNodeStorage).removeJobNodeIfExisted("execution/2/completed"); verify(jobNodeStorage).isJobNodeExisted("leader/execution/necessary"); verify(configService).getShardingTotalCount(); verify(jobNodeStorage).removeJobNodeIfExisted("execution/2"); verify(jobNodeStorage).removeJobNodeIfExisted("leader/execution/necessary"); verify(jobNodeStorage).removeJobNodeIfExisted("leader/execution/cleaning"); verify(jobNodeStorage).isJobNodeExisted("leader/execution/cleaning"); }
@Test public void assertCleanPreviousExecutionInfoWhenNeedFixExecutionInfoForNewValuesEqual() { when(jobNodeStorage.isJobNodeExisted("execution")).thenReturn(true); when(leaderElectionService.isLeader()).thenReturn(true); when(jobNodeStorage.getJobNodeChildrenKeys("execution")).thenReturn(Arrays.asList("0", "1", "2")); when(jobNodeStorage.isJobNodeExisted("leader/execution/necessary")).thenReturn(true); when(configService.getShardingTotalCount()).thenReturn(3); executionService.cleanPreviousExecutionInfo(); verify(jobNodeStorage).isJobNodeExisted("execution"); verify(leaderElectionService).isLeader(); verify(jobNodeStorage).fillEphemeralJobNode("leader/execution/cleaning", ""); verify(jobNodeStorage).getJobNodeChildrenKeys("execution"); verify(jobNodeStorage).removeJobNodeIfExisted("execution/0/completed"); verify(jobNodeStorage).removeJobNodeIfExisted("execution/1/completed"); verify(jobNodeStorage).removeJobNodeIfExisted("execution/2/completed"); verify(jobNodeStorage).isJobNodeExisted("leader/execution/necessary"); verify(configService).getShardingTotalCount(); verify(jobNodeStorage).removeJobNodeIfExisted("leader/execution/necessary"); verify(jobNodeStorage).removeJobNodeIfExisted("leader/execution/cleaning"); verify(jobNodeStorage).isJobNodeExisted("leader/execution/cleaning"); }