ExecutionContextService { public JobExecutionMultipleShardingContext getJobExecutionShardingContext(final List<Integer> shardingItems) { JobExecutionMultipleShardingContext result = new JobExecutionMultipleShardingContext(); result.setJobName(jobConfiguration.getJobName()); result.setShardingTotalCount(configService.getShardingTotalCount()); result.setShardingItems(shardingItems); boolean isMonitorExecution = configService.isMonitorExecution(); if (isMonitorExecution) { removeRunningItems(shardingItems); } result.setJobParameter(configService.getJobParameter()); result.setMonitorExecution(isMonitorExecution); result.setFetchDataCount(configService.getFetchDataCount()); if (result.getShardingItems().isEmpty()) { return result; } Map<Integer, String> shardingItemParameters = configService.getShardingItemParameters(); for (int each : result.getShardingItems()) { if (shardingItemParameters.containsKey(each)) { result.getShardingItemParameters().put(each, shardingItemParameters.get(each)); } } result.setOffsets(offsetService.getOffsets(result.getShardingItems())); return result; } ExecutionContextService(final CoordinatorRegistryCenter coordinatorRegistryCenter, final JobConfiguration jobConfiguration); JobExecutionMultipleShardingContext getJobExecutionShardingContext(final List<Integer> shardingItems);  }
@Test public void assertGetJobExecutionShardingContextWhenNotAssignShardingItem() { when(configService.getShardingTotalCount()).thenReturn(3); when(configService.isFailover()).thenReturn(false); when(configService.isMonitorExecution()).thenReturn(false); when(configService.getFetchDataCount()).thenReturn(10); JobExecutionMultipleShardingContext expected = new JobExecutionMultipleShardingContext(); expected.setJobName("testJob"); expected.setShardingTotalCount(3); expected.setFetchDataCount(10); assertThat(executionContextService.getJobExecutionShardingContext(Collections.<Integer>emptyList()), new ReflectionEquals(expected)); verify(configService).getShardingTotalCount(); verify(configService).isMonitorExecution(); verify(configService).getFetchDataCount(); }
@Test public void assertGetJobExecutionShardingContextWhenAssignShardingItems() { when(configService.getShardingTotalCount()).thenReturn(3); when(configService.isFailover()).thenReturn(false); when(configService.isMonitorExecution()).thenReturn(false); when(configService.getFetchDataCount()).thenReturn(10); Map<Integer, String> shardingItemParameters = new HashMap<>(3); shardingItemParameters.put(0, "A"); shardingItemParameters.put(1, "B"); shardingItemParameters.put(2, "C"); when(configService.getShardingItemParameters()).thenReturn(shardingItemParameters); Map<Integer, String> offsets = new HashMap<>(2); offsets.put(0, "offset0"); offsets.put(1, "offset1"); when(offsetService.getOffsets(Arrays.asList(0, 1))).thenReturn(offsets); JobExecutionMultipleShardingContext expected = new JobExecutionMultipleShardingContext(); expected.setJobName("testJob"); expected.setShardingTotalCount(3); expected.setFetchDataCount(10); expected.setShardingItems(Arrays.asList(0, 1)); expected.getShardingItemParameters().put(0, "A"); expected.getShardingItemParameters().put(1, "B"); expected.setOffsets(offsets); assertThat(executionContextService.getJobExecutionShardingContext(Arrays.asList(0, 1)), new ReflectionEquals(expected)); verify(configService).getShardingTotalCount(); verify(configService).isMonitorExecution(); verify(configService).getFetchDataCount(); verify(configService).getShardingItemParameters(); verify(offsetService).getOffsets(Arrays.asList(0, 1)); }
@Test public void assertGetJobExecutionShardingContextWhenHasRunningItems() { when(configService.getShardingTotalCount()).thenReturn(3); when(configService.isFailover()).thenReturn(true); when(configService.isMonitorExecution()).thenReturn(true); when(jobNodeStorage.isJobNodeExisted("execution/0/running")).thenReturn(false); when(jobNodeStorage.isJobNodeExisted("execution/1/running")).thenReturn(true); when(configService.getFetchDataCount()).thenReturn(10); Map<Integer, String> shardingItemParameters = new HashMap<>(3); shardingItemParameters.put(0, "A"); shardingItemParameters.put(1, "B"); shardingItemParameters.put(2, "C"); when(configService.getShardingItemParameters()).thenReturn(shardingItemParameters); Map<Integer, String> offsets = new HashMap<>(1); offsets.put(0, "offset0"); when(offsetService.getOffsets(Collections.singletonList(0))).thenReturn(offsets); JobExecutionMultipleShardingContext expected = new JobExecutionMultipleShardingContext(); expected.setJobName("testJob"); expected.setShardingTotalCount(3); expected.setFetchDataCount(10); expected.setShardingItems(Collections.singletonList(0)); expected.getShardingItemParameters().put(0, "A"); expected.setMonitorExecution(true); expected.setOffsets(offsets); assertThat(executionContextService.getJobExecutionShardingContext(Lists.newArrayList(0, 1)), new ReflectionEquals(expected)); verify(configService).getShardingTotalCount(); verify(configService).isMonitorExecution(); verify(jobNodeStorage).isJobNodeExisted("execution/0/running"); verify(jobNodeStorage).isJobNodeExisted("execution/1/running"); verify(configService).getFetchDataCount(); verify(configService).getShardingItemParameters(); verify(offsetService).getOffsets(Collections.singletonList(0)); }
@Test public void assertGetJobExecutionShardingContextWhenAssignShardingItems() { when(configService.getShardingTotalCount()).thenReturn(3); when(configService.isFailover()).thenReturn(false); when(configService.isMonitorExecution()).thenReturn(false); when(configService.getFetchDataCount()).thenReturn(10); Map<Integer, String> shardingItemParameters = new HashMap<Integer, String>(3); shardingItemParameters.put(0, "A"); shardingItemParameters.put(1, "B"); shardingItemParameters.put(2, "C"); when(configService.getShardingItemParameters()).thenReturn(shardingItemParameters); Map<Integer, String> offsets = new HashMap<Integer, String>(2); offsets.put(0, "offset0"); offsets.put(1, "offset1"); when(offsetService.getOffsets(Arrays.asList(0, 1))).thenReturn(offsets); JobExecutionMultipleShardingContext expected = new JobExecutionMultipleShardingContext(); expected.setJobName("testJob"); expected.setShardingTotalCount(3); expected.setFetchDataCount(10); expected.setShardingItems(Arrays.asList(0, 1)); expected.getShardingItemParameters().put(0, "A"); expected.getShardingItemParameters().put(1, "B"); expected.setOffsets(offsets); assertThat(executionContextService.getJobExecutionShardingContext(Arrays.asList(0, 1)), new ReflectionEquals(expected)); verify(configService).getShardingTotalCount(); verify(configService).isMonitorExecution(); verify(configService).getFetchDataCount(); verify(configService).getShardingItemParameters(); verify(offsetService).getOffsets(Arrays.asList(0, 1)); }
@Test public void assertGetJobExecutionShardingContextWhenHasRunningItems() { when(configService.getShardingTotalCount()).thenReturn(3); when(configService.isFailover()).thenReturn(true); when(configService.isMonitorExecution()).thenReturn(true); when(jobNodeStorage.isJobNodeExisted("execution/0/running")).thenReturn(false); when(jobNodeStorage.isJobNodeExisted("execution/1/running")).thenReturn(true); when(configService.getFetchDataCount()).thenReturn(10); Map<Integer, String> shardingItemParameters = new HashMap<Integer, String>(3); shardingItemParameters.put(0, "A"); shardingItemParameters.put(1, "B"); shardingItemParameters.put(2, "C"); when(configService.getShardingItemParameters()).thenReturn(shardingItemParameters); Map<Integer, String> offsets = new HashMap<Integer, String>(1); offsets.put(0, "offset0"); when(offsetService.getOffsets(Collections.singletonList(0))).thenReturn(offsets); JobExecutionMultipleShardingContext expected = new JobExecutionMultipleShardingContext(); expected.setJobName("testJob"); expected.setShardingTotalCount(3); expected.setFetchDataCount(10); expected.setShardingItems(Collections.singletonList(0)); expected.getShardingItemParameters().put(0, "A"); expected.setMonitorExecution(true); expected.setOffsets(offsets); assertThat(executionContextService.getJobExecutionShardingContext(Lists.newArrayList(0, 1)), new ReflectionEquals(expected)); verify(configService).getShardingTotalCount(); verify(configService).isMonitorExecution(); verify(jobNodeStorage).isJobNodeExisted("execution/0/running"); verify(jobNodeStorage).isJobNodeExisted("execution/1/running"); verify(configService).getFetchDataCount(); verify(configService).getShardingItemParameters(); verify(offsetService).getOffsets(Collections.singletonList(0)); }