Thread interruption flag leaks from BoundedElasticThreadPerTaskScheduler thread into user code

[petkov]

When virtual threads are enabled with reactor.schedulers.defaultBoundedElasticOnVirtualThreads property, BoundedElasticThreadPerTaskScheduler thread leaks interrupted signal into user code. Below is the test that shows that:

    @Test
    void shouldOnErrorResumeLambdaBeExecutedInsideUninterruptedThread() {
        System.setProperty("reactor.schedulers.defaultBoundedElasticOnVirtualThreads", "true");
        Mono.fromCallable(() -> {
                throw new RuntimeException("$ERROR$");
            })
            .subscribeOn(Schedulers.boundedElastic())
            .onErrorResume(e -> {
                Assertions.assertFalse(Thread.interrupted()); // The test fails here
                return Mono.empty();
            })
            .block();
    }

Expected Behavior

The lambda which is called by onErrorResume should be run on a thread that is not interrupted.

Actual Behavior

The lambda which is called by onErrorResume is run on an interrupted Thread.

Reactor Core version: 3.6.6

Notes

Bellow test passes if an exception is thrown inside map function:

    @Test
    void shouldOnErrorResumeLambdaBeExecutedInsideUninterruptedThread() {
        System.setProperty("reactor.schedulers.defaultBoundedElasticOnVirtualThreads", "true");
        Mono.fromCallable(() -> {
                return "NOT_EMPTY";
            })
            .map(s -> {
                throw new RuntimeException("$ERROR$");
            })
            .subscribeOn(Schedulers.boundedElastic())
            .onErrorResume(e -> {
                Assertions.assertFalse(Thread.interrupted()); // The test fails here
                return Mono.empty();
            })
            .block();
    }

[chemicL]

Thank you for the report. It indeed looks like a bug, I'll try to review the implementation and figure out what to do.

[kwondh5217]

Hi @chemicL !

I might be missing something here, but It seems that the interrupted flag persists because the thread isn’t in a blocking state, which prevents the flag from resetting.

In the first test, the interruption occurs before the error handler (onErrorResume) is executed, so the interrupted flag leaks into user code. However, in the second test, the error handler (onErrorResume) executes before the interrupt is triggered, which allows the test case to pass as the interrupted flag hasn't been set yet.

I was wondering if this approach might help:
SchedulerTask.dispose() in BoundedElasticThreadPerTaskScheduler

if (isRunning(previousState)) {
 if (hasFuture(previousState)) {
  this.scheduledFuture.cancel(true);
 }
 if (this.carrier.getState() == Thread.State.WAITING ||
   this.carrier.getState() == Thread.State.TIMED_WAITING ||
   this.carrier.getState() == Thread.State.BLOCKED) {
  this.carrier.interrupt();
 }
 return;
}

After this change, the test cases pass as expected. It ensures that the interrupt is only triggered when the thread is in a waiting or blocked state, potentially avoiding unnecessary leaks into user code.

What do you think about this adjustment? I'm open to feedback if I misunderstood anything! 🙇