This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new df18224bb [runtime] Reject RuntimeExecutionMode.BATCH with the default
batch state backend (#1104)
df18224bb is described below
commit df18224bb9cedd5e37d8c694d29e058354301d06
Author: Ashfaq <[email protected]>
AuthorDate: Mon Sep 14 09:32:20 2026 +0530
[runtime] Reject RuntimeExecutionMode.BATCH with the default batch state
backend (#1104)
Generated-by: Claude Code 2.1.226 (Claude Sonnet 5)
---
.../apache/flink/agents/runtime/CompileUtils.java | 41 ++++++++++++++++++++++
.../flink/agents/runtime/CompileUtilsTest.java | 34 ++++++++++++++++++
2 files changed, 75 insertions(+)
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/CompileUtils.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/CompileUtils.java
index c1bf0805c..65e998cb8 100644
--- a/runtime/src/main/java/org/apache/flink/agents/runtime/CompileUtils.java
+++ b/runtime/src/main/java/org/apache/flink/agents/runtime/CompileUtils.java
@@ -22,15 +22,19 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.agents.plan.AgentPlan;
import org.apache.flink.agents.runtime.operator.ActionExecutionOperatorFactory;
+import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.configuration.ExecutionOptions;
+import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.typeinfo.python.PickledByteArrayTypeInfo;
import org.apache.flink.types.Row;
import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkState;
/** A utility class that bridges Flink DataStream/SQL with the Flink Agents
agent. */
public class CompileUtils {
@@ -97,6 +101,7 @@ public class CompileUtils {
TypeInformation<OUT> outTypeInformation,
boolean inputIsJava,
boolean pythonKeyIsPickled) {
+ checkBatchStateBackendCompatibility(keyedInputStream);
return (DataStream<OUT>)
keyedInputStream
.transform(
@@ -107,6 +112,42 @@ public class CompileUtils {
.setParallelism(keyedInputStream.getParallelism());
}
+ /**
+ * Rejects a job configuration known to silently drop records: explicit
{@code
+ * RuntimeExecutionMode.BATCH} together with Flink's batch-specific
keyed-state backend (enabled
+ * by default in that mode).
+ *
+ * <p>{@link ActionExecutionOperatorFactory}'s operator keeps pending
action tasks in keyed
+ * state across mailbox continuations that can span multiple keys of the
same subtask. Flink's
+ * batch keyed-state backend assumes a key is fully processed before
moving to the next one and
+ * clears keyed state on every key switch, so a continuation for one key
can be silently
+ * discarded by the operator moving on to process another key first. The
job still reports
+ * {@code FINISHED}; only the affected records go missing. See
+ * https://github.com/apache/flink-agents/issues/939.
+ *
+ * <p>This only catches an explicitly configured {@code BATCH} runtime
mode. {@code
+ * RuntimeExecutionMode.AUTOMATIC} (the default) that later resolves to
batch execution because
+ * every source happens to be bounded is not detected here, since that
resolution has not
+ * happened yet at graph-construction time.
+ */
+ private static void checkBatchStateBackendCompatibility(KeyedStream<?, ?>
keyedInputStream) {
+ ReadableConfig configuration =
+ keyedInputStream.getExecutionEnvironment().getConfiguration();
+ boolean isExplicitBatchMode =
+ configuration.get(ExecutionOptions.RUNTIME_MODE) ==
RuntimeExecutionMode.BATCH;
+ boolean usesBatchStateBackend =
configuration.get(ExecutionOptions.USE_BATCH_STATE_BACKEND);
+ checkState(
+ !isExplicitBatchMode || !usesBatchStateBackend,
+ "Flink Agents does not support RuntimeExecutionMode.BATCH with
Flink's batch"
+ + " keyed-state backend
(execution.batch-state-backend.enabled, enabled by"
+ + " default in BATCH mode): pending action-task state
can be silently"
+ + " discarded when the operator processes another key
before a key's"
+ + " workflow finishes, causing records to be dropped
without any error."
+ + " Either run this job with
RuntimeExecutionMode.STREAMING, or set"
+ + " execution.batch-state-backend.enabled to false.
See"
+ + " https://github.com/apache/flink-agents/issues/939
for details.");
+ }
+
/** Returns whether a PyFlink Row field uses its default pickle
representation. */
static boolean isPickledPythonFieldType(TypeInformation<?>
typeInformation, int fieldIndex) {
checkArgument(fieldIndex >= 0, "Field index must not be negative, but
got %s", fieldIndex);
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/CompileUtilsTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/CompileUtilsTest.java
index 96a0ce8fc..b2dba2ea2 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/CompileUtilsTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/CompileUtilsTest.java
@@ -20,9 +20,12 @@ package org.apache.flink.agents.runtime;
import com.fasterxml.jackson.core.JsonProcessingException;
import org.apache.flink.agents.plan.AgentPlan;
import org.apache.flink.agents.runtime.operator.ActionExecutionOperatorTest;
+import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ExecutionOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
@@ -39,6 +42,8 @@ import java.util.List;
import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link CompileUtils}. */
@@ -103,6 +108,35 @@ public class CompileUtilsTest {
checkResult(resultList);
}
+ @Test
+ void rejectsExplicitBatchModeWithDefaultBatchStateBackend() {
+ // execution.batch-state-backend.enabled defaults to true in BATCH
mode, which is exactly
+ // the combination that silently drops records (issue #939).
+ Configuration conf = new Configuration();
+ conf.set(ExecutionOptions.RUNTIME_MODE, RuntimeExecutionMode.BATCH);
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(conf);
+ KeyedStream<Long, Long> keyedInputStream =
env.fromData(testSequence).keyBy(x -> x);
+
+ assertThatIllegalStateException()
+ .isThrownBy(() ->
CompileUtils.connectToAgent(keyedInputStream, TEST_AGENT_PLAN))
+ .withMessageContaining("RuntimeExecutionMode.BATCH")
+
.withMessageContaining("execution.batch-state-backend.enabled");
+ }
+
+ @Test
+ void allowsExplicitBatchModeWithBatchStateBackendDisabled() {
+ // The documented workaround: disabling the batch-specific state
backend restores the
+ // per-key-processed-to-completion semantics the operator relies on.
+ Configuration conf = new Configuration();
+ conf.set(ExecutionOptions.RUNTIME_MODE, RuntimeExecutionMode.BATCH);
+ conf.set(ExecutionOptions.USE_BATCH_STATE_BACKEND, false);
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(conf);
+ KeyedStream<Long, Long> keyedInputStream =
env.fromData(testSequence).keyBy(x -> x);
+
+ assertThatCode(() -> CompileUtils.connectToAgent(keyedInputStream,
TEST_AGENT_PLAN))
+ .doesNotThrowAnyException();
+ }
+
@Test
void detectsPickledAndExplicitByteArrayPythonKeyTypes() {
assertThat(
