AsyncJobLauncherImpl implements AsyncJobLauncher,
                                  InitializingBean { @Override public void executeJob(final String jobSequenceId) { Assert.notNull(jobSequenceId); try { taskPoolLimit.acquire(); } catch (InterruptedException e) { LOGGER.error(LogId.EAL025054, e, jobSequenceId); throw new BatchException(e); } if (!asyncJobWorker.beforeExecute(jobSequenceId)) { taskPoolLimit.release(); LOGGER.info(LogId.IAL025021, jobSequenceId); return; } try { threadPoolTaskExecutor.execute(new Runnable() { @Override public void run() { try { asyncJobWorker.executeWorker(jobSequenceId); } catch (Throwable t) { LOGGER.error(LogId.EAL025053, t); } finally { taskPoolLimit.release(); } } }); } catch (TaskRejectedException e) { LOGGER.error(LogId.EAL025047, e, jobSequenceId); taskPoolLimit.release(); } } AsyncJobLauncherImpl(ThreadPoolTaskExecutor threadPoolTaskExecutor,
            AsyncJobWorker asyncJobWorker); void setFair(boolean fair); @Override void executeJob(final String jobSequenceId); @Override void shutdown(); @Override void afterPropertiesSet();  }
@Test public void testExecuteJob01() throws Exception { AsyncJobLauncherImpl asyncJobLauncher = new AsyncJobLauncherImpl( threadPoolTaskExecutor, asyncJobWorker); try { asyncJobLauncher.executeJob(null); fail(); } catch (IllegalArgumentException e) { assertEquals( "[Assertion failed] - this argument is required; it must not be null", e.getMessage()); } }
@Test public void testExecuteJob02() throws Exception { Semaphore semaphore = new Semaphore(10); doReturn(false).when(asyncJobWorker).beforeExecute(anyString()); AsyncJobLauncherImpl asyncJobLauncher = new AsyncJobLauncherImpl( threadPoolTaskExecutor, asyncJobWorker); asyncJobLauncher.taskPoolLimit = semaphore; asyncJobLauncher.executeJob("0000000001"); assertEquals(10, semaphore.availablePermits()); assertThat(logger.getLoggingEvents(), is(asList(info( "[IAL025021] Skipped this job execution because this job has already been started by another. jobSequenceId:0000000001")))); verify(threadPoolTaskExecutor, never()).execute(any(Runnable.class)); verify(asyncJobWorker, never()).executeWorker(anyString()); }
@Test public void testExecuteJob03() throws Exception { Semaphore semaphore = new Semaphore(10); doThrow(TaskRejectedException.class).when(threadPoolTaskExecutor) .execute(any(Runnable.class)); doReturn(true).when(asyncJobWorker).beforeExecute(anyString()); AsyncJobLauncherImpl asyncJobLauncher = new AsyncJobLauncherImpl( threadPoolTaskExecutor, asyncJobWorker); asyncJobLauncher.taskPoolLimit = semaphore; asyncJobLauncher.executeJob("0000000001"); verify(asyncJobWorker, never()).executeWorker(anyString()); assertEquals(10, semaphore.availablePermits()); assertTrue(logger.getLoggingEvents() .get(0) .getThrowable() .get() instanceof TaskRejectedException); assertEquals( "[EAL025047] This job cannot be accepted for execution. jobSequenceId:0000000001", logger.getLoggingEvents().get(0).getMessage()); }
@Test public void testExecuteJob04() throws Exception { Semaphore semaphore = spy(new Semaphore(10)); doThrow(InterruptedException.class).when(semaphore).acquire(); doReturn(true).when(asyncJobWorker).beforeExecute(anyString()); AsyncJobLauncherImpl asyncJobLauncher = new AsyncJobLauncherImpl( threadPoolTaskExecutor, asyncJobWorker); asyncJobLauncher.taskPoolLimit = semaphore; try { asyncJobLauncher.executeJob("0000000001"); fail("An exception has not been occured."); } catch (BatchException be) { assertTrue(be.getCause() instanceof InterruptedException); verify(asyncJobWorker, never()).executeWorker(anyString()); assertTrue(logger.getLoggingEvents() .get(0) .getThrowable() .get() instanceof InterruptedException); assertEquals( "[EAL025054] When the job is waiting for a available worker thread, it has been interrupted. jobSequenceId:0000000001", logger.getLoggingEvents().get(0).getMessage()); } }
@Test public void testExecuteJob05() throws Exception { doReturn(true).when(asyncJobWorker).beforeExecute(anyString()); doNothing().when(asyncJobWorker).executeWorker(anyString()); Semaphore semaphore = new Semaphore(10); doAnswer(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocationOnMock) throws Throwable { Runnable runnable = invocationOnMock.getArgumentAt(0, Runnable.class); runnable.run(); return null; } }).when(threadPoolTaskExecutor).execute(any(Runnable.class)); AsyncJobLauncherImpl asyncJobLauncher = new AsyncJobLauncherImpl( threadPoolTaskExecutor, asyncJobWorker); asyncJobLauncher.taskPoolLimit = semaphore; asyncJobLauncher.executeJob("0000000001"); verify(asyncJobWorker).executeWorker("0000000001"); assertEquals(10, semaphore.availablePermits()); }
@Test public void testExecuteJob07() throws Exception { final RuntimeException runtimeException = new RuntimeException( "exception in AsyncJobWorker#executeWorker()"); doReturn(true).when(asyncJobWorker).beforeExecute(anyString()); doThrow(runtimeException).when(asyncJobWorker) .executeWorker("0000000001"); Semaphore semaphore = new Semaphore(10); doAnswer(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocation) throws Throwable { Runnable runnable = invocation.getArgumentAt(0, Runnable.class); runnable.run(); return null; } }).when(threadPoolTaskExecutor).execute(any(Runnable.class)); AsyncJobLauncherImpl asyncJobLauncher = new AsyncJobLauncherImpl( threadPoolTaskExecutor, asyncJobWorker); asyncJobLauncher.taskPoolLimit = semaphore; asyncJobLauncher.executeJob("0000000001"); assertEquals(10, asyncJobLauncher.taskPoolLimit.availablePermits()); verify(threadPoolTaskExecutor).execute(any(Runnable.class)); }
@Test public void testExecuteJob09() throws Exception { final RuntimeException runtimeException = new RuntimeException( "exception in AsyncJobWorker#executeWorker()"); doReturn(true).when(asyncJobWorker).beforeExecute(anyString()); doThrow(runtimeException).when(asyncJobWorker) .executeWorker("0000000001"); Semaphore semaphore = new Semaphore(10); doAnswer(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocationOnMock) throws Throwable { Runnable runnable = invocationOnMock.getArgumentAt(0, Runnable.class); runnable.run(); return null; } }).when(threadPoolTaskExecutor).execute(any(Runnable.class)); AsyncJobLauncherImpl asyncJobLauncher = new AsyncJobLauncherImpl( threadPoolTaskExecutor, asyncJobWorker); asyncJobLauncher.taskPoolLimit = semaphore; asyncJobLauncher.executeJob("0000000001"); verify(asyncJobWorker).executeWorker("0000000001"); assertEquals(10, semaphore.availablePermits()); assertThat(logger.getLoggingEvents(), is(asList(error(runtimeException, "[EAL025053] An exception occurred. please see below the stacktrace.")))); }