[Bug] Bounded workflows silently drop records in batch mode with multiple keys
- Dominant language
- Java
- Stars
- 452
- Forks
- 167
- Avg merge
- 5d 9h
- Merged PRs (30d)
- 49
Description
### 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!
Contributor guide
Research direction
Start with ActionExecutionOperator and its pending action keyed-state handling and mailbox continuation scheduling. Reproduce the issue in a real BATCH execution environment with parallelism 1, five differently keyed inputs, and synchronous stages. Done means every input completes its workflow and produces output, with a regression test covering multiple keys.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, python
- Domain
- backend, distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 48/100