Work inside the handler invocations and Context.run() calls is not interrupted when invocations are aborted

Inizia confrontando l’implementazione Kotlin HandlerRunner con la gestione di onClosedInvocationStreamHook di Java HandlerRunner. Segui il percorso dell’executor fino a ContextImpl e le chiamate a Context.run(), quindi verifica che il lavoro in esecuzione negli handler venga interrotto dopo la chiusura di un invocation stream dovuta a un abort. L’issue non indica file o test specifici.

Hey there. I was testing restate sdk using java and noticed that work inside the handlers was still being performed long after restate server aborted the invocations (abort timeout reached). I noticed that HandlerRunner for kotlin has an implementation that cancels work using onClosedInvocationStreamHook, but java implementation for HandlerRunner ignores onClosedInvocationStreamHook. Is this intentional?

If not, 2 solutions come to mind:

1. Keep a collection of threads which are doing work by registering thread's reference inside the executor calls:

**initialize worker thread references collection here**
**set onClosedInvocationStreamHook to iterate over the worker threads and interrupt them here**
options.executor.execute(
                () -> {
                  HANDLER_CONTEXT_THREAD_LOCAL.set(handlerContext);
                  try (Scope ignored =
                      handlerContext.request().openTelemetryContext().makeCurrent()) {
                    **add thread's ref to collection here**
                    runnable.run();
                  } finally {
                    **remove thread's ref from collection here**
                    HANDLER_CONTEXT_THREAD_LOCAL.remove();
                  }
                });

2. Require ExecutorService instead of Executor when setting up the handler options, so that ExecutorService.submit(...) method is used instead of Executor.execute(...) and the output Futures can be collected and used to interrupt the worker threads when onClosedInvocationStreamHook is called (Future has cancel method which can interrupt thread). However, I see that Executor is passed down to ContextImpl so this change could propagate beyond HandlerRunner's scope. Maybe an adapter between ExecutorService and Executor can be made to limit the scope (ExecutorService wrapped by Executor which is passed into ContextImpl, when Executor.execute is called, the call is delegated to ExecutorService.submit and the resulting Futures are collected inside HandlerRunner)