[runtime] Reject RuntimeExecutionMode.BATCH with the default batch state backend

[Ashfaqbs]

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 [Bug] Bounded workflows silently drop records in batch mode with multiple keys #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.

Verification evidence

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.

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

• doc-not-needed

Was this patch authored or co-authored using generative AI tooling?

• Yes

Generated-by: Claude Code 2.1.226 (Claude Sonnet 5)