sangkyoonnam opened a new issue, #1294: URL: https://github.com/apache/flink-agents/issues/1294
### Search before asking - [x] I searched in the [issues](https://github.com/apache/flink-agents/issues) and found nothing similar. ### Description Canceling a job on the parallel execution engine while an action waits in `ContinuationActionExecutor.executeAsync` can make a worker thread throw an uncaught `IllegalStateException`, which Flink logs as an ERROR through `ClusterUncaughtExceptionHandler`. I reproduced it by canceling the Java Quickstart `WorkflowSingleAgentExample` during a pending chat request. The job still reached CANCELED in those runs. ``` ERROR org.apache.flink.runtime.util.ClusterUncaughtExceptionHandler [] - WARNING: Thread 'flink-agents-java-async-pool-5-thread-2' produced an uncaught exception. ... java.lang.IllegalStateException: Current thread does not own the lock. owner=null, current=Thread[#88,flink-agents-java-async-pool-5-thread-2,5,Flink Task Threads] at org.apache.flink.agents.runtime.operator.parallel.ParallelExecutionLock.checkReentrant(ParallelExecutionLock.java:186) at org.apache.flink.agents.runtime.operator.parallel.ParallelExecutionLock.releaseToMain(ParallelExecutionLock.java:168) at org.apache.flink.agents.runtime.operator.parallel.ParallelExecutionCoordinator.runOneTask(ParallelExecutionCoordinator.java:213) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144) ... ``` From the code, this is the path I think produces it. On the parallel engine, `executeAsync` releases the lock before running the supplier, then calls `acquireByWorker` and restores the runner context in a `finally`. During cancellation, the coordinator's `shutdown()` calls `shutdownNow()`, which interrupts the workers. If that re-acquire throws `InterruptedException`, the context isn't restored, and `work.execute()` records the failure and returns with the worker no longer owning the lock. `schedule.markDone(node)` then fails its ownership check, and the coordinator's catch block tries to dispatch that exception through `mailboxDispatcher`. The `finally` in `runOneTask` then calls `releaseToMain()`, and that ownership check throws the exception that escapes the pool thread. An interrupt during the initial `acquireByWorker` in `runOneTask` returns before the inner `try`/`finally`, so it doesn't explain this stack. `executeAllAsync` re-acquires the same way. Guarding only `releaseToMain()` would remove the logged exception, but `markDone()` would still run without ownership. The directions I can see are handling, in the coordinator, a worker that comes back from `execute()` without the lock, or keeping ownership and context intact in the executor when the re-acquire is interrupted, without risking an indefinite wait during shutdown. cc @eemario for context on #1147, and @pltbkd for the related lifecycle work in #1218. I can send a PR with a regression test in `ParallelExecutionCoordinatorTest` that uses the base `ContinuationActionExecutor` and shuts the coordinator down while a worker is re-acquiring the lock. Would a separate PR for this cancellation path fit with the lifecycle work in #1218? ### How to reproduce 1. Build main with JDK 17 (`./tools/build.sh -j`) and copy the full `dist/flink-2.3/target/flink-agents-dist-flink-2.3-0.4-SNAPSHOT.jar` (not `-thin`) into `$FLINK_HOME/lib`. Start a Flink 2.3.0 session cluster and Ollama with `qwen3:8b`. 2. Submit `org.apache.flink.agents.examples.WorkflowSingleAgentExample` from `examples/target/flink-agents-examples-0.4-SNAPSHOT.jar`. Its actions are all Java, the base executor has no continuation support, and `parallel-execution.enabled` defaults to true, so it runs on the parallel engine. 3. Cancel the job while a chat request is pending. I canceled after the first result appeared in the TaskManager output. 4. Check the TaskManager log for the ERROR above. How many appear depends on the timing of the cancel. ### Version and environment flink-agents main at `a50a8a18` and `742faf60`; the parallel engine and executor code is unchanged on the current main (`b6702839`). Flink 2.3.0 standalone session cluster, macOS 15.3.1 on Apple M4. Seen on JDK 17.0.14 (12 occurrences across several canceled runs) and on JDK 21.0.11 (2 in one cancel). The JDK 21 run used a JDK 17-built distribution without the JDK 21 executor (#1292), so it also ran on the parallel engine. I haven't investigated effects beyond the uncaught exception and the ERROR log. ### Are you willing to submit a PR? - [x] I'm willing to submit a PR! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
