JobFacade { public JobExecutionMultipleShardingContext getShardingContext() { boolean isFailover = configService.isFailover(); if (isFailover) { List<Integer> failoverItems = failoverService.getLocalHostFailoverItems(); if (!failoverItems.isEmpty()) { return executionContextService.getJobExecutionShardingContext(failoverItems); } } shardingService.shardingIfNecessary(); List<Integer> shardingItems = shardingService.getLocalHostShardingItems(); if (isFailover) { shardingItems.removeAll(failoverService.getLocalHostTakeOffItems()); } return executionContextService.getJobExecutionShardingContext(shardingItems); } JobFacade(final CoordinatorRegistryCenter coordinatorRegistryCenter, final JobConfiguration jobConfiguration, final List<ElasticJobListener> elasticJobListeners); String getJobName(); int getConcurrentDataProcessThreadCount(); void checkMaxTimeDiffSecondsTolerable(); void failoverIfNecessary(); void registerJobBegin(final JobExecutionMultipleShardingContext shardingContext); void registerJobCompleted(final JobExecutionMultipleShardingContext shardingContext, boolean completeFlag); JobExecutionMultipleShardingContext getShardingContext(); boolean misfireIfNecessary(final List<Integer> shardingItems); void clearMisfire(final List<Integer> shardingItems); boolean isExecuteMisfired(final List<Integer> shardingItems); boolean isEligibleForJobRunning(); boolean isNeedSharding(); void updateOffset(final int item, final String offset); void cleanPreviousExecutionInfo(); void beforeJobExecuted(final JobExecutionMultipleShardingContext shardingContext); void afterJobExecuted(final JobExecutionMultipleShardingContext shardingContext); boolean inSkipTime(); boolean isLocalServerDisabled();  }
@Test public void testGetShardingContextWhenIsFailoverEnableAndFailover() { JobExecutionMultipleShardingContext shardingContext = new JobExecutionMultipleShardingContext(); when(configService.isFailover()).thenReturn(true); when(failoverService.getLocalHostFailoverItems()).thenReturn(Collections.singletonList(1)); when(executionContextService.getJobExecutionShardingContext(Collections.singletonList(1))).thenReturn(shardingContext); assertThat(jobFacade.getShardingContext(), is(shardingContext)); verify(shardingService, times(0)).shardingIfNecessary(); }
@Test public void testGetShardingContextWhenIsFailoverEnableAndNotFailover() { JobExecutionMultipleShardingContext shardingContext = new JobExecutionMultipleShardingContext(); when(configService.isFailover()).thenReturn(true); when(failoverService.getLocalHostFailoverItems()).thenReturn(Collections.<Integer>emptyList()); when(shardingService.getLocalHostShardingItems()).thenReturn(Lists.newArrayList(0, 1)); when(failoverService.getLocalHostTakeOffItems()).thenReturn(Collections.singletonList(0)); when(executionContextService.getJobExecutionShardingContext(Collections.singletonList(1))).thenReturn(shardingContext); assertThat(jobFacade.getShardingContext(), is(shardingContext)); verify(shardingService).shardingIfNecessary(); }
@Test public void testGetShardingContextWhenIsFailoverDisable() { JobExecutionMultipleShardingContext shardingContext = new JobExecutionMultipleShardingContext(); when(configService.isFailover()).thenReturn(false); when(shardingService.getLocalHostShardingItems()).thenReturn(Lists.newArrayList(0, 1)); when(executionContextService.getJobExecutionShardingContext(Lists.newArrayList(0, 1))).thenReturn(shardingContext); assertThat(jobFacade.getShardingContext(), is(shardingContext)); verify(shardingService).shardingIfNecessary(); }