Repository navigation
[Bug] Bounded workflows silently drop records in batch mode with multiple keys #939
Description
Search before asking
- I searched in the 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
-
Create a Python workflow with multiple synchronous stages:
InputEvent -> Stage1Event -> Stage2Event -> Stage3Event -> OutputEvent -
Run it with:
RuntimeExecutionMode.BATCH- parallelism
1 - five input records
- a different key for every record
-
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!
Activity
weiqingy commentedon Aug 1, 2026
Thanks for the detailed repro. Confirmed the mechanism on current
main, not just 0.2.1 / Flink 1.20.3:actionTasksKStateandpendingInputEventsKStateare keyed state, whilecurrentProcessingKeysOpStateis operator state. That is why the job still reports FINISHED:endInputwaits on the operator state that survived, while the per-key task lists are already gone.mainbuilds against,BatchExecutionKeyedStateBackend.setCurrentKeystill clears every registered state on any key change, so this is not specific to 1.20.x.processEventsubmits a mail,processActionTaskForKeystarts withsetCurrentKey(key), and an unfinished task submits another mail.Agreed that
enabled=falseis an escape hatch, not a fix. The option defaults totrue(ExecutionOptions.USE_BATCH_STATE_BACKEND), so a plain BATCH job hits this with no configuration at all, and that default path is the one that has to work.Two ways I can see to get there:
processElement, which is what the batch backend assumes callers do anyway.waitInFlightEventsFinishedalready does this atendInputby spinning onmailboxExecutor.yield().I lean toward the first. The second does not stop at those two lists:
sequenceNumberKState,sensoryMemStateandshortTermMemStateare keyed too, so preserving cross-key overlap means moving all per-key runtime state off the batch backend, which is close to reimplementingenabled=falseinside the operator. And the overlap that draining gives up is not somethingenabled=truecan support anyway.One open question either way: the behavior has to be conditional on the batch backend, and
getKeyedStateBackend() instanceof BatchExecutionKeyedStateBackendis the signal I found. Is there a cleaner one?Batch also has no coverage today, every
set_runtime_modein the e2e suites passesSTREAMING, so the regression test you described would be the first of its kind. Glad to take this on.WDYT?
wellkilo commentedon Sep 28, 2026
I'd like to take this on and implement the underlying runtime fix, if nobody is actively working on it.
I checked current
mainand the merged #1104 safeguard. The intended scope is:ActionExecutionOperatorat runtime, so explicitBATCHandAUTOMATICmodes are handled according to the backend Flink actually created.processElementreturns. This keeps all keyed state for one key alive until the batch backend's key-complete contract is satisfied.The public
CompileUtils.connectToAgentsignatures and event contracts would remain unchanged. The externally visible change is that the currently rejected configuration completes normally without dropping records.@weiqingy, I noticed your earlier "glad to take this on" comment. Please let me know if you already have an active implementation; otherwise I will proceed with the above focused scope.
weiqingy commentedon Sep 29, 2026
@wellkilo I haven’t received any reply from @Sxnan yet, so I haven’t started. Please feel free to go ahead and take it. @Sxnan can chime in with more context.
[runtime] Finish batch backend support and regression coverage
[runtime] Finish batch backend support and regression coverage
[runtime][java] Complete batch workflows before key switches (#1172)