Hey Flink community, Im currently exploring async io API in Flink. I notice timeout doesn’t get triggered when i use the flink’s Executors.directExecutor() as recommended here

package github.jhchee.functions;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.flink.util.concurrent.Executors;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
//import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
public class DelayAsyncFunction extends RichAsyncFunction<Integer, String> {

    private transient Executor executor;

    @Override
    public void open(Configuration parameters) {
        <http://log.info|log.info>("Initializing executor");
        executor = Executors.directExecutor();
        // executor = Executors.newFixedThreadPool(10); // this works
    }

    @Override
    public void asyncInvoke(Integer input, ResultFuture<String> resultFuture) {
        CompletableFuture.supplyAsync(() -> performAsyncOperation(input), executor)
                         .thenAccept(result -> resultFuture.complete(Collections.singletonList(result)));
    }

    @Override
    public void timeout(Integer input, ResultFuture<String> resultFuture) {
        String message = String.format("Failed to delay for %d seconds", input);
        resultFuture.complete(Collections.singletonList(message));
    }

    private String performAsyncOperation(Integer input) {
        try {
            TimeUnit.SECONDS.sleep(input);
        } catch (InterruptedException ignored) {

        }
        // Perform the asynchronous operation and return the result
        return String.format("Delay for %d seconds", input);
    }

    @Override
    public void close() {
        ((ExecutorService) executor).shutdown();
    }
}