apache / apache/flink-agents

[Bug] Bounded workflows silently drop records in batch mode with multiple keys

Open
#939 1 comment 0 reactions 0 assignees View on GitHub
bug priority/major
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.