@Test public void assertRegisterJobBeginWithoutNextFireTime() { when(configService.isMonitorExecution()).thenReturn(true); when(jobScheduleController.getNextFireTime()).thenReturn(null); JobRegistry.getInstance().addJobScheduleController("testJob", jobScheduleController); JobExecutionMultipleShardingContext jobExecutionShardingContext = new JobExecutionMultipleShardingContext(); jobExecutionShardingContext.setShardingItems(Arrays.asList(0, 1, 2)); executionService.registerJobBegin(jobExecutionShardingContext); verify(configService).isMonitorExecution(); verify(serverService).updateServerStatus(ServerStatus.RUNNING); verify(jobNodeStorage).fillEphemeralJobNode("execution/0/running", ""); verify(jobNodeStorage).fillEphemeralJobNode("execution/1/running", ""); verify(jobNodeStorage).fillEphemeralJobNode("execution/2/running", ""); verify(jobNodeStorage).replaceJobNode(eq("execution/0/lastBeginTime"), anyLong()); verify(jobNodeStorage).replaceJobNode(eq("execution/1/lastBeginTime"), anyLong()); verify(jobNodeStorage).replaceJobNode(eq("execution/2/lastBeginTime"), anyLong()); }
public void registerJobBegin(final JobExecutionMultipleShardingContext jobExecutionShardingContext) { if (!jobExecutionShardingContext.getShardingItems().isEmpty() && configService.isMonitorExecution()) { serverService.updateServerStatus(ServerStatus.RUNNING); for (int each : jobExecutionShardingContext.getShardingItems()) { jobNodeStorage.fillEphemeralJobNode(ExecutionNode.getRunningNode(each), ""); if (!localHostService.getIp().equals(jobNodeStorage.getJobNodeData(ExecutionNode.getServerIp(each)))) { jobNodeStorage.replaceJobNode(ExecutionNode.getServerIp(each), localHostService.getIp()); } JobScheduleController jobScheduleController = JobRegistry.getInstance().getJobScheduleController(jobConfiguration.getJobName()); if (null != jobScheduleController) { Date nextFireTime = jobScheduleController.getNextFireTime(); if (null != nextFireTime) { jobNodeStorage.replaceJobNode(ExecutionNode.getNextFireTimeNode(each), nextFireTime.getTime()); } } jobNodeStorage.replaceJobNode(ExecutionNode.getLastBeginTimeNode(each), System.currentTimeMillis()); } } }
ExecutionService { public void registerJobBegin(final JobExecutionMultipleShardingContext jobExecutionShardingContext) { if (!jobExecutionShardingContext.getShardingItems().isEmpty() && configService.isMonitorExecution()) { serverService.updateServerStatus(ServerStatus.RUNNING); for (int each : jobExecutionShardingContext.getShardingItems()) { jobNodeStorage.fillEphemeralJobNode(ExecutionNode.getRunningNode(each), ""); if (!localHostService.getIp().equals(jobNodeStorage.getJobNodeData(ExecutionNode.getServerIp(each)))) { jobNodeStorage.replaceJobNode(ExecutionNode.getServerIp(each), localHostService.getIp()); } JobScheduleController jobScheduleController = JobRegistry.getInstance().getJobScheduleController(jobConfiguration.getJobName()); if (null != jobScheduleController) { Date nextFireTime = jobScheduleController.getNextFireTime(); if (null != nextFireTime) { jobNodeStorage.replaceJobNode(ExecutionNode.getNextFireTimeNode(each), nextFireTime.getTime()); } } jobNodeStorage.replaceJobNode(ExecutionNode.getLastBeginTimeNode(each), System.currentTimeMillis()); } } } }
ExecutionService { public void registerJobBegin(final JobExecutionMultipleShardingContext jobExecutionShardingContext) { if (!jobExecutionShardingContext.getShardingItems().isEmpty() && configService.isMonitorExecution()) { serverService.updateServerStatus(ServerStatus.RUNNING); for (int each : jobExecutionShardingContext.getShardingItems()) { jobNodeStorage.fillEphemeralJobNode(ExecutionNode.getRunningNode(each), ""); if (!localHostService.getIp().equals(jobNodeStorage.getJobNodeData(ExecutionNode.getServerIp(each)))) { jobNodeStorage.replaceJobNode(ExecutionNode.getServerIp(each), localHostService.getIp()); } JobScheduleController jobScheduleController = JobRegistry.getInstance().getJobScheduleController(jobConfiguration.getJobName()); if (null != jobScheduleController) { Date nextFireTime = jobScheduleController.getNextFireTime(); if (null != nextFireTime) { jobNodeStorage.replaceJobNode(ExecutionNode.getNextFireTimeNode(each), nextFireTime.getTime()); } } jobNodeStorage.replaceJobNode(ExecutionNode.getLastBeginTimeNode(each), System.currentTimeMillis()); } } } ExecutionService(final CoordinatorRegistryCenter coordinatorRegistryCenter, final JobConfiguration jobConfiguration); }
ExecutionService { public void registerJobBegin(final JobExecutionMultipleShardingContext jobExecutionShardingContext) { if (!jobExecutionShardingContext.getShardingItems().isEmpty() && configService.isMonitorExecution()) { serverService.updateServerStatus(ServerStatus.RUNNING); for (int each : jobExecutionShardingContext.getShardingItems()) { jobNodeStorage.fillEphemeralJobNode(ExecutionNode.getRunningNode(each), ""); if (!localHostService.getIp().equals(jobNodeStorage.getJobNodeData(ExecutionNode.getServerIp(each)))) { jobNodeStorage.replaceJobNode(ExecutionNode.getServerIp(each), localHostService.getIp()); } JobScheduleController jobScheduleController = JobRegistry.getInstance().getJobScheduleController(jobConfiguration.getJobName()); if (null != jobScheduleController) { Date nextFireTime = jobScheduleController.getNextFireTime(); if (null != nextFireTime) { jobNodeStorage.replaceJobNode(ExecutionNode.getNextFireTimeNode(each), nextFireTime.getTime()); } } jobNodeStorage.replaceJobNode(ExecutionNode.getLastBeginTimeNode(each), System.currentTimeMillis()); } } } ExecutionService(final CoordinatorRegistryCenter coordinatorRegistryCenter, final JobConfiguration jobConfiguration); void registerJobBegin(final JobExecutionMultipleShardingContext jobExecutionShardingContext); void cleanPreviousExecutionInfo(); void registerJobCompleted(final JobExecutionMultipleShardingContext jobExecutionShardingContext, boolean completeFlag); 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(); }
ExecutionService { public void registerJobBegin(final JobExecutionMultipleShardingContext jobExecutionShardingContext) { if (!jobExecutionShardingContext.getShardingItems().isEmpty() && configService.isMonitorExecution()) { serverService.updateServerStatus(ServerStatus.RUNNING); for (int each : jobExecutionShardingContext.getShardingItems()) { jobNodeStorage.fillEphemeralJobNode(ExecutionNode.getRunningNode(each), ""); if (!localHostService.getIp().equals(jobNodeStorage.getJobNodeData(ExecutionNode.getServerIp(each)))) { jobNodeStorage.replaceJobNode(ExecutionNode.getServerIp(each), localHostService.getIp()); } JobScheduleController jobScheduleController = JobRegistry.getInstance().getJobScheduleController(jobConfiguration.getJobName()); if (null != jobScheduleController) { Date nextFireTime = jobScheduleController.getNextFireTime(); if (null != nextFireTime) { jobNodeStorage.replaceJobNode(ExecutionNode.getNextFireTimeNode(each), nextFireTime.getTime()); } } jobNodeStorage.replaceJobNode(ExecutionNode.getLastBeginTimeNode(each), System.currentTimeMillis()); } } } ExecutionService(final CoordinatorRegistryCenter coordinatorRegistryCenter, final JobConfiguration jobConfiguration); void registerJobBegin(final JobExecutionMultipleShardingContext jobExecutionShardingContext); void cleanPreviousExecutionInfo(); void registerJobCompleted(final JobExecutionMultipleShardingContext jobExecutionShardingContext, boolean completeFlag); 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();  }