Sxnan opened a new issue, #939: URL: https://github.com/apache/flink-agents/issues/939
### Search before asking - [x] I searched in the [issues](https://github.com/apache/flink-agents/issues) and found nothing similar. ### Description When a Flink Agents workflow runs with `RuntimeExecutionMode.BATCH`, records may be silently dropped if one operator subtask processes multiple keys. The job finishes successfully without errors, but usually only the last key processed by each subtask produces output. `ActionExecutionOperator` stores pending action tasks in keyed state while scheduling workflow continuations through the mailbox. Flink's batch keyed-state backend assumes that a key is fully processed before moving to the next key and clears keyed state whenever the current key changes. Therefore, when the operator processes key B before the mailbox continuation for key A runs, A's pending tasks are cleared. Switching back to A can also clear B's tasks. Expected: every input record completes its workflow and produces output. Actual: pending workflows are silently discarded while the Flink job reports `FINISHED`. Setting `execution.batch-state-backend.enabled=false` avoids the issue. ### Possible fix In batch mode, the operator must finish the complete action chain for the current key before allowing input processing to move to another key. Alternatively, Flink Agents jobs should disable the batch keyed-state backend. A regression test should use a real BATCH execution environment with at least two keys assigned to the same subtask. ### How to reproduce 1. Create a Python workflow with multiple synchronous stages: `InputEvent -> Stage1Event -> Stage2Event -> Stage3Event -> OutputEvent` 2. Run it with: - `RuntimeExecutionMode.BATCH` - parallelism `1` - five input records - a different key for every record 3. Observe that only one record reaches the output: `5 inputs -> 1 output` Control results: - Using one fixed key: `5 inputs -> 5 outputs` - Setting `execution.batch-state-backend.enabled=false`: `5 inputs -> 5 outputs` - The issue remains reproducible when all action delays are removed. ### Version and environment - Flink Agents: 0.2.1 - Apache Flink: 1.20.3 - Python: 3.11 - Java: OpenJDK 11 - Runtime mode: BATCH - OS: macOS ARM64 The issue was reproduced locally with the official `flink-agents==0.2.1` package and does not require a vendor-specific Flink distribution. ### Are you willing to submit a PR? - [ ] 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]
