DefaultJobExecutor implements JobExecutor, Initializable, Disposable { @Override public Job execute(String jobType, Request request) throws JobException { Job job = createJob(jobType, request); execute(job); return job; }  @Override void initialize(); @Override void dispose(); @Override Job getCurrentJob(JobGroupPath path); @Override Job getJob(List<String> id); @Override Job execute(String jobType, Request request); @Override void execute(Job job);  }
@Test void matchingGroupPathAreBlocked() { GroupedJobInitializer groupedJobInitializer = mock(GroupedJobInitializer.class); when(groupedJobInitializer.getPoolSize()).thenReturn(1); when(groupedJobInitializer.getDefaultPriority()).thenReturn(Thread.NORM_PRIORITY); when(this.groupedJobInitializerManager.getGroupedJobInitializer(any())).thenReturn(groupedJobInitializer); TestBasicGroupedJob jobA = groupedJob("A"); TestBasicGroupedJob jobAB = groupedJob("A", "B"); TestBasicGroupedJob job12 = groupedJob("1", "2"); TestBasicGroupedJob job1 = groupedJob("1"); jobA.lock(); jobAB.lock(); job12.lock(); job1.lock(); this.executor.execute(jobA); this.executor.execute(job12); waitJobWaiting(jobA); waitJobWaiting(job12); this.executor.execute(jobAB); this.executor.execute(job1); assertSame(State.WAITING, jobA.getStatus().getState()); assertNull(jobAB.getStatus().getState()); jobA.unlock(); waitJobWaiting(jobAB); assertSame(State.FINISHED, jobA.getStatus().getState()); assertSame(State.WAITING, jobAB.getStatus().getState()); jobAB.unlock(); waitJobFinished(jobAB); assertSame(State.FINISHED, jobA.getStatus().getState()); assertSame(State.FINISHED, jobAB.getStatus().getState()); assertSame(State.WAITING, job12.getStatus().getState()); assertNull(job1.getStatus().getState()); job12.unlock(); waitJobWaiting(job1); assertSame(State.FINISHED, job12.getStatus().getState()); assertSame(State.WAITING, job1.getStatus().getState()); job1.unlock(); waitJobFinished(job1); assertSame(State.FINISHED, job1.getStatus().getState()); assertSame(State.FINISHED, job1.getStatus().getState()); }
@Test void matchingGroupPathAreBlockedPoolMultiSizeParentFirst() { GroupedJobInitializer groupedJobInitializer = mock(GroupedJobInitializer.class); when(groupedJobInitializer.getPoolSize()).thenReturn(1); when(groupedJobInitializer.getDefaultPriority()).thenReturn(Thread.NORM_PRIORITY); JobGroupPath jobGroupPathA = new JobGroupPath(Collections.singletonList("A")); when(this.groupedJobInitializerManager.getGroupedJobInitializer(eq(jobGroupPathA))) .thenReturn(groupedJobInitializer); groupedJobInitializer = mock(GroupedJobInitializer.class); when(groupedJobInitializer.getPoolSize()).thenReturn(2); when(groupedJobInitializer.getDefaultPriority()).thenReturn(Thread.NORM_PRIORITY); JobGroupPath jobGroupPathAB = new JobGroupPath(Arrays.asList("A", "B")); when(this.groupedJobInitializerManager.getGroupedJobInitializer(eq(jobGroupPathAB))) .thenReturn(groupedJobInitializer); TestBasicGroupedJob jobA1 = groupedJob("A"); TestBasicGroupedJob jobA2 = groupedJob("A"); TestBasicGroupedJob jobAB1 = groupedJob("A", "B"); TestBasicGroupedJob jobAB2 = groupedJob("A", "B"); TestBasicGroupedJob jobAB3 = groupedJob("A", "B"); jobA1.lock(); jobA2.lock(); jobAB1.lock(); jobAB2.lock(); jobAB3.lock(); this.executor.execute(jobA1); waitJobWaiting(jobA1); this.executor.execute(jobAB1); this.executor.execute(jobAB2); assertSame(State.WAITING, jobA1.getStatus().getState()); assertNull(jobA2.getStatus().getState()); assertNull(jobAB1.getStatus().getState()); assertNull(jobAB2.getStatus().getState()); jobA1.unlock(); waitJobFinished(jobA1); waitJobWaiting(jobAB1); waitJobWaiting(jobAB2); this.executor.execute(jobA2); assertSame(State.FINISHED, jobA1.getStatus().getState()); assertSame(State.WAITING, jobAB1.getStatus().getState()); assertSame(State.WAITING, jobAB2.getStatus().getState()); assertNull(jobA2.getStatus().getState()); jobAB1.unlock(); waitJobFinished(jobAB1); this.executor.execute(jobAB3); assertSame(State.FINISHED, jobAB1.getStatus().getState()); assertSame(State.WAITING, jobAB2.getStatus().getState()); assertNull(jobAB3.getStatus().getState()); assertNull(jobA2.getStatus().getState()); jobAB2.unlock(); waitJobFinished(jobAB2); waitJobWaiting(jobA2); assertSame(State.FINISHED, jobAB2.getStatus().getState()); assertSame(State.WAITING, jobA2.getStatus().getState()); assertNull(jobAB3.getStatus().getState()); jobA2.unlock(); waitJobFinished(jobA2); waitJobWaiting(jobAB3); assertSame(State.FINISHED, jobA2.getStatus().getState()); assertSame(State.WAITING, jobAB3.getStatus().getState()); jobAB3.unlock(); waitJobFinished(jobAB3); assertSame(State.FINISHED, jobAB3.getStatus().getState()); }
@Test void matchingGroupPathAreBlockedPoolMultiSizeChildrenFirst() throws InterruptedException { GroupedJobInitializer groupedJobInitializer = mock(GroupedJobInitializer.class); when(groupedJobInitializer.getPoolSize()).thenReturn(2); when(groupedJobInitializer.getDefaultPriority()).thenReturn(Thread.NORM_PRIORITY); JobGroupPath jobGroupPathA = new JobGroupPath(Collections.singletonList("A")); when(this.groupedJobInitializerManager.getGroupedJobInitializer(eq(jobGroupPathA))) .thenReturn(groupedJobInitializer); groupedJobInitializer = mock(GroupedJobInitializer.class); when(groupedJobInitializer.getPoolSize()).thenReturn(2); when(groupedJobInitializer.getDefaultPriority()).thenReturn(Thread.NORM_PRIORITY); JobGroupPath jobGroupPathAB = new JobGroupPath(Arrays.asList("A", "B")); when(this.groupedJobInitializerManager.getGroupedJobInitializer(eq(jobGroupPathAB))) .thenReturn(groupedJobInitializer); TestBasicGroupedJob jobA1 = groupedJob("A"); TestBasicGroupedJob jobA2 = groupedJob("A"); TestBasicGroupedJob jobAB1 = groupedJob("A", "B"); TestBasicGroupedJob jobAB2 = groupedJob("A", "B"); TestBasicGroupedJob jobAB3 = groupedJob("A", "B"); jobA1.lock(); jobA2.lock(); jobAB1.lock(); jobAB2.lock(); jobAB3.lock(); this.executor.execute(jobAB1); this.executor.execute(jobAB2); waitJobWaiting(jobAB1); waitJobWaiting(jobAB2); this.executor.execute(jobA1); Thread.sleep(WAIT_VALUE); this.executor.execute(jobA2); assertSame(State.WAITING, jobAB1.getStatus().getState()); assertSame(State.WAITING, jobAB2.getStatus().getState()); assertNull(jobA1.getStatus().getState()); assertNull(jobA2.getStatus().getState()); jobAB1.unlock(); waitJobFinished(jobAB1); waitJobWaiting(jobA1); this.executor.execute(jobAB3); assertSame(State.FINISHED, jobAB1.getStatus().getState()); assertSame(State.WAITING, jobAB2.getStatus().getState()); assertSame(State.WAITING, jobA1.getStatus().getState()); assertNull(jobA2.getStatus().getState()); assertNull(jobAB3.getStatus().getState()); jobAB2.unlock(); waitJobFinished(jobAB2); waitJobWaiting(jobA2); assertSame(State.FINISHED, jobAB2.getStatus().getState()); assertSame(State.WAITING, jobA1.getStatus().getState()); assertSame(State.WAITING, jobA2.getStatus().getState()); assertNull(jobAB3.getStatus().getState()); jobA1.unlock(); jobA2.unlock(); waitJobFinished(jobA1); waitJobFinished(jobA2); waitJobWaiting(jobAB3); assertSame(State.FINISHED, jobA1.getStatus().getState()); assertSame(State.FINISHED, jobA2.getStatus().getState()); assertSame(State.WAITING, jobAB3.getStatus().getState()); jobAB3.unlock(); waitJobFinished(jobAB3); assertSame(State.FINISHED, jobAB3.getStatus().getState()); }