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

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)

Start by comparing the Kotlin HandlerRunner implementation with the Java HandlerRunner handling of onClosedInvocationStreamHook. Trace the executor path into ContextImpl and the Context.run() calls, then verify that work running in handlers is interrupted after an invocation stream closes due to an abort. The issue does not name specific files or tests.