SimpleAsyncTaskScheduler: Returned ScheduledFuture does not track provided task execution

[Sheikah45]

Affects: Spring Framework 6.1.5

The ScheduledFuture returned by SimpleAsyncTaskScheduler does not track the execution state of the submitted task, but rather tracks the execution state of the scheduling of the task. This is a noted difference from the ThreadPoolTaskScheduler where the returned ScheduledFuture tracks the execution of the submitted task itself.

This has consequences for anyone migrating their taskScheduler to use virtual threads and currently interact with the returned future, since SimpleAsyncTaskScheduler is the recommended class to do so.

From my investigation this is caused by the wrapping of the provided task into a delegate task which submits the provided task to the underlying SimpleAsyncTaskExecutor execute method. This delegation effectively hides the execution of the provided task from the ScheduledExecutorService meaning once the task is submitted to the underlying execute method it is no longer able to track the status of the task execution and sets its state to success.

Below I have included some test cases that demonstrate the effects of this in both the successful and exceptional cases. They can also be found in the following repo https://github.com/Sheikah45/TaskSchedulerExample

If it is not possible to allow the delegating task status to be tracked I would suggest adding a note to the documentation of the SimpleAsyncTaskScheduler to make it clear what the expected behavior of the returned Scheduled future is, since as it stands now the behavior appears to contradict the documentation of the schedule methods which state: Returns: a ScheduledFuture representing pending completion of the task

    @Test
    void tracksSchedulingOfTask() throws Exception {
        AtomicBoolean atomicBoolean = new AtomicBoolean(false);
        SimpleAsyncTaskScheduler taskScheduler = new SimpleAsyncTaskScheduler();
        ScheduledFuture<?> future = taskScheduler.schedule(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            atomicBoolean.set(true);
        }, Instant.now().plusSeconds(1));

        future.get();

        assertTrue(atomicBoolean.get());
    }

    @Test
    void tracksExecutionTask() throws Exception {
        AtomicBoolean atomicBoolean = new AtomicBoolean(false);
        ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
        taskScheduler.initialize();
        ScheduledFuture<?> future = taskScheduler.schedule(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            atomicBoolean.set(true);
        }, Instant.now().plusSeconds(1));

        future.get();

        assertTrue(atomicBoolean.get());
    }

    @Test
    void doesNotThrowTaskException() throws Exception {
        SimpleAsyncTaskScheduler taskScheduler = new SimpleAsyncTaskScheduler();
        ScheduledFuture<?> schedule = taskScheduler.schedule(() -> {
            throw new RuntimeException();
        }, Instant.now().plusSeconds(1));

        assertThrows(ExecutionException.class, schedule::get);
    }

    @Test
    void throwsTaskException() throws Exception {
        ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
        taskScheduler.initialize();
        ScheduledFuture<?> schedule = taskScheduler.schedule(() -> {
            throw new RuntimeException();
        }, Instant.now().plusSeconds(1));

        assertThrows(ExecutionException.class, schedule::get);
    }