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

まず、Kotlin HandlerRunner の実装と、Java HandlerRunner における onClosedInvocationStreamHook の処理を比較します。executor から ContextImpl へ至る経路と Context.run() の呼び出しを追跡し、その後、abort によって invocation stream が閉じられた後に、handler で実行中の処理が中断されることを確認します。この issue では具体的なファイルやテストは指定されていません。

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)