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.