Ashfaqbs opened a new pull request, #1104: URL: https://github.com/apache/flink-agents/pull/1104
Linked issue: #939 ### Purpose of change A job configured with `RuntimeExecutionMode.BATCH` and Flink's batch keyed-state backend (`execution.batch-state-backend.enabled`, enabled by default in `BATCH` mode) now fails immediately at job-graph construction with a clear error, instead of running to completion and silently dropping records. `ActionExecutionOperator` keeps pending action tasks in keyed state across mailbox continuations that can span multiple keys of the same subtask. The batch keyed-state backend assumes a key is fully processed before the operator moves to the next one and clears keyed state on every key switch, so a continuation for one key can be discarded when the operator processes another key first — the job still reports `FINISHED`; only the affected records go missing. **Runtime flow:** `CompileUtils.connectToAgent` (both the Python and Java entry points funnel into the same private overload) now calls `checkBatchStateBackendCompatibility` before constructing the `ActionExecutionOperator`. That reads `ExecutionOptions.RUNTIME_MODE` and `ExecutionOptions.USE_BATCH_STATE_BACKEND` off `keyedInputStream.getExecutionEnvironment().getConfiguration()` and throws before the operator is ever wired into the graph. **Key decisions:** - Fail fast at graph construction rather than attempt the deeper fix (making the operator safe under batch key-switch semantics), which is a separate, larger change out of scope here. - Only an explicitly configured `BATCH` mode is detected. `RuntimeExecutionMode.AUTOMATIC` (the default) that later resolves to batch execution because every source happens to be bounded is not caught, since that resolution hasn't happened yet at graph-construction time — inspecting source boundedness to predict `AUTOMATIC`'s eventual resolution was rejected as disproportionate for a case the explicit, documented `BATCH` setting already covers. ### Behavioral Semantics #### Interaction decisions | `RUNTIME_MODE` | `USE_BATCH_STATE_BACKEND` | Behavior | |---|---|---| | `STREAMING` | `true` / `false` | Unaffected — check only fires for `BATCH` | | `BATCH` | `true` (default) | Rejected: `IllegalStateException` before the operator is constructed | | `BATCH` | `false` | Allowed — the documented workaround | | `AUTOMATIC` | `true` / `false` | Unaffected — not detected (see Key decisions) | #### Behavioral contracts 1. `connectToAgent` throws `IllegalStateException` before constructing the operator when `RUNTIME_MODE=BATCH` and `USE_BATCH_STATE_BACKEND=true`. 2. `connectToAgent` behaves exactly as before this change for every other `RUNTIME_MODE`/`USE_BATCH_STATE_BACKEND` combination. 3. The exception message names both workarounds (`RuntimeExecutionMode.STREAMING`, or setting `execution.batch-state-backend.enabled=false`) and links to #939. #### Failure behavior The only new failure path: an unsupported `BATCH` + batch-state-backend-enabled configuration raises `IllegalStateException` synchronously, before any operator is created — not a retry, not silently absorbed. For the Python entry point, this check runs after the existing pickled-input-type and agent-plan-JSON validations, so it doesn't change their ordering or messages. No other failure path changes. ### Tests | Contract | Test | |---|---| | 1 | `rejectsExplicitBatchModeWithDefaultBatchStateBackend` | | 2 (BATCH + disabled) | `allowsExplicitBatchModeWithBatchStateBackendDisabled` | | 2 (STREAMING, unaffected) | pre-existing `testJavaNoKeyedStreamConnectToAgent` / `testJavaKeyedStreamConnectToAgent` | | 3 | `rejectsExplicitBatchModeWithDefaultBatchStateBackend` asserts the message contains both `RuntimeExecutionMode.BATCH` and `execution.batch-state-backend.enabled` | Not verified: `AUTOMATIC` mode resolving to batch execution via all-bounded sources; an actual executed `BATCH` job reproducing the original multi-key data-loss scenario (would need a heavier MiniCluster run) — the causal chain for *why* the collision happens relies on the issue's own trace of Flink's batch state-backend semantics, not a reproduction inside this PR. <details> <summary>Verification evidence</summary> Ran `CompileUtilsTest` (9/9 pass) plus the full `operator`, `context`, and `skill` packages after a clean rebuild (`mvn -pl runtime clean`, then `mvn -pl api,plan -am install -DskipTests` to pick up unrelated upstream API changes already on `main`). 249 tests, 7 pre-existing failures unrelated to this change: 4 are a Windows-only temp-directory-deletion quirk in `ClasspathSkillRepositoryTest` (locked JAR files), and 3 (`FileSystemSkillRepositoryTest`, `SkillManagerTest` x2) pick up this machine's own local Claude Code skills directory contents (`github`, `nano-banana-pro`) rather than the repo's test fixtures — both reproduce identically on unmodified `main` in this environment. `mvn -pl runtime spotless:check` clean. </details> ### API No public API changes. `connectToAgent`'s signature and behavior for every previously-supported configuration are unchanged; this adds a new, additive rejection for a configuration that was previously accepted but silently broken. ### Documentation - [x] `doc-not-needed` ### Was this patch authored or co-authored using generative AI tooling? - [x] Yes Generated-by: Claude Code 2.1.226 (Claude Sonnet 5) -- 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]
