TaskletExecutionService { CompletableFuture<Void> beginExecute( @Nonnull List<? extends Tasklet> tasklets, @Nonnull CompletableFuture<Void> cancellationFuture, @Nonnull ClassLoader jobClassLoader ) { final ExecutionTracker executionTracker = new ExecutionTracker(tasklets.size(), cancellationFuture); try { final Map<Boolean, List<Tasklet>> byCooperation = tasklets.stream().collect(partitioningBy(Tasklet::isCooperative)); submitCooperativeTasklets(executionTracker, jobClassLoader, byCooperation.get(true)); submitBlockingTasklets(executionTracker, jobClassLoader, byCooperation.get(false)); } catch (Throwable t) { executionTracker.future.internalCompleteExceptionally(t); } return executionTracker.future; } TaskletExecutionService(NodeEngineImpl nodeEngine, int threadCount, HazelcastProperties properties); void shutdown(); void awaitWorkerTermination(); static final String TASKLET_INIT_CLOSE_EXECUTOR_NAME; }
@Test public void when_nonBlockingTaskletIsCancelled_then_completesEarly() throws Exception { final List<MockTasklet> tasklets = Stream.generate(() -> new MockTasklet().callsBeforeDone(Integer.MAX_VALUE)) .limit(100).collect(toList()); CompletableFuture<Void> f = tes.beginExecute(tasklets, cancellationFuture, classLoaderMock); cancellationFuture.cancel(true); tasklets.forEach(MockTasklet::assertNotDone); exceptionRule.expect(CancellationException.class); f.get(); }
@Test public void when_blockingTaskletIsCancelled_then_completeEarly() throws ExecutionException, InterruptedException { final List<MockTasklet> tasklets = Stream.generate(() -> new MockTasklet().blocking().callsBeforeDone(Integer.MAX_VALUE)) .limit(100).collect(toList()); CompletableFuture<Void> f = tes.beginExecute(tasklets, cancellationFuture, classLoaderMock); cancellationFuture.cancel(true); tasklets.forEach(MockTasklet::assertNotDone); exceptionRule.expect(CancellationException.class); f.get(); }
@Test public void when_blockingSleepingTaskletIsCancelled_then_completeEarly() throws Exception { final List<MockTasklet> tasklets = Stream.generate(() -> new MockTasklet().sleeping().callsBeforeDone(Integer.MAX_VALUE)) .limit(100).collect(toList()); CompletableFuture<Void> f = tes.beginExecute(tasklets, cancellationFuture, classLoaderMock); cancellationFuture.cancel(true); tasklets.forEach(MockTasklet::assertNotDone); assertTrueEventually(f::isDone); exceptionRule.expect(CancellationException.class); cancellationFuture.get(); }
@Test public void when_nonBlockingCancelled_then_doneCallBackFiredAfterActualDone() throws Exception { CountDownLatch proceedLatch = new CountDownLatch(1); final List<MockTasklet> tasklets = Stream.generate(() -> new MockTasklet().waitOnLatch(proceedLatch).callsBeforeDone(Integer.MAX_VALUE)) .limit(100).collect(toList()); CompletableFuture<Void> f = tes.beginExecute(tasklets, cancellationFuture, classLoaderMock); cancellationFuture.cancel(true); assertFalse("future should not be completed until tasklets are completed.", f.isDone()); proceedLatch.countDown(); assertTrueEventually(() -> { assertTrue("future should be completed eventually", f.isDone()); }); exceptionRule.expect(CancellationException.class); cancellationFuture.get(); }
@Test public void when_twoNonBlockingTasklets_then_differentWorker() { TaskletAssertingThreadLocal t1 = new TaskletAssertingThreadLocal(); TaskletAssertingThreadLocal t2 = new TaskletAssertingThreadLocal(); assertTrue(t1.isCooperative()); CompletableFuture<Void> f1 = tes.beginExecute(singletonList(t1), new CompletableFuture<>(), classLoaderMock); CompletableFuture<Void> f2 = tes.beginExecute(singletonList(t2), new CompletableFuture<>(), classLoaderMock); f1.join(); f2.join(); }
@Test public void when_tryCompleteOnReturnedFuture_then_fails() { final MockTasklet t = new MockTasklet().callsBeforeDone(Integer.MAX_VALUE); CompletableFuture<Void> f = tes.beginExecute(singletonList(t), cancellationFuture, classLoaderMock); exceptionRule.expect(UnsupportedOperationException.class); f.complete(null); }
@Test public void when_tryCompleteExceptionallyOnReturnedFuture_then_fails() { final MockTasklet t = new MockTasklet().callsBeforeDone(Integer.MAX_VALUE); CompletableFuture<Void> f = tes.beginExecute(singletonList(t), cancellationFuture, classLoaderMock); exceptionRule.expect(UnsupportedOperationException.class); f.completeExceptionally(new RuntimeException()); }
@Test public void when_tryCancelOnReturnedFuture_then_fails() { final MockTasklet t = new MockTasklet().callsBeforeDone(Integer.MAX_VALUE); CompletableFuture<Void> f = tes.beginExecute(singletonList(t), cancellationFuture, classLoaderMock); exceptionRule.expect(UnsupportedOperationException.class); f.cancel(true); }
@Test public void when_cancellationFutureCompleted_then_fails() throws Throwable { final MockTasklet t = new MockTasklet().callsBeforeDone(Integer.MAX_VALUE); CompletableFuture<Void> f = tes.beginExecute(singletonList(t), cancellationFuture, classLoaderMock); cancellationFuture.complete(null); exceptionRule.expect(IllegalStateException.class); try { f.join(); } catch (CompletionException e) { throw peel(e); } }