You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by we...@apache.org on 2026/07/15 03:49:54 UTC
(flink-agents) branch release-0.3 updated (86ec7052 -> ea245810)
This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a change to branch release-0.3
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
from 86ec7052 [python] Remove PyFlink gateway fallback from AgentConfigOptions (#867)
new c649ebe7 [dist] Add Flink 2.3 distribution and make it the default (#856)
new 09b81387 [hotfix] Propagate action task Errors from mailbox (#880)
new 06866c56 [runtime] Binary serde for MemoryUpdate values via versioned Kryo envelope (#874)
new 77aedb9c [tools] Update install.sh for Flink Agents 0.3.0 and Python 3.12 (#888)
new ea245810 [tools] Keep install plan edits consistent (#899)
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/workflows/ci.yml | 6 +
dist/flink-2.2/pom.xml | 44 ++++
dist/{flink-2.2 => flink-2.3}/pom.xml | 4 +-
dist/pom.xml | 2 +
.../docs/development/memory/long_term_memory.md | 2 +-
docs/content/docs/faq/faq.md | 2 +-
.../pom.xml | 14 +-
pom.xml | 2 +-
python/_build_backend/tests/test_backend.py | 7 +
python/jar_manifest.json | 6 +
.../runtime/actionstate/ActionStateSerde.java | 119 ++++++++++
.../runtime/operator/ActionExecutionOperator.java | 6 +-
.../runtime/actionstate/ActionStateSerdeTest.java | 137 ++++++++++++
.../operator/ActionExecutionOperatorTest.java | 58 +++++
tools/e2e.sh | 2 +-
tools/install.sh | 180 +++++++++++++--
tools/test/helpers/load.bash | 7 +-
tools/test/integration/dry_run_extra.bats | 2 +-
tools/test/unit/edit_plan_quote.bats | 3 +
tools/test/unit/reconcile_plan_after_edit.bats | 55 +++++
tools/test/unit/revalidate_python_constraint.bats | 244 +++++++++++++++++++++
tools/test/unit/validate_python_bin.bats | 39 +++-
tools/ut.sh | 4 +-
23 files changed, 911 insertions(+), 34 deletions(-)
copy dist/{flink-2.2 => flink-2.3}/pom.xml (97%)
create mode 100644 tools/test/unit/reconcile_plan_after_edit.bats
create mode 100644 tools/test/unit/revalidate_python_constraint.bats
(flink-agents) 02/05: [hotfix] Propagate action task Errors from mailbox (#880)
Posted by we...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch release-0.3
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
commit 09b81387d62e73a89ebcb4ea9a28362d5ee49315
Author: Wenjin Xie <16...@users.noreply.github.com>
AuthorDate: Wed Jul 8 15:08:10 2026 +0800
[hotfix] Propagate action task Errors from mailbox (#880)
Catch Throwable in mailbox-submitted action task processing so Errors are rethrown through the mailbox and fail the task visibly instead of remaining in a Future while processing keys stay in-flight.
Co-authored-by: Codex <co...@openai.com>
(cherry picked from commit 588016170be176dab269721bfcd64bbe4c6351e3)
---
.../runtime/operator/ActionExecutionOperator.java | 6 ++-
.../operator/ActionExecutionOperatorTest.java | 58 ++++++++++++++++++++++
2 files changed, 62 insertions(+), 2 deletions(-)
diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java
index 665551bb..46d09e0a 100644
--- a/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java
+++ b/runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java
@@ -283,12 +283,14 @@ public class ActionExecutionOperator<IN, OUT> extends AbstractStreamOperator<OUT
private void tryProcessActionTaskForKey(Object key) {
try {
processActionTaskForKey(key);
- } catch (Exception e) {
+ } catch (Throwable t) {
+ // MailboxExecutor.submit() stores task failures in its Future. Catch Throwable and
+ // rethrow via execute() so Errors fail the task instead of leaving the key in-flight.
mailboxExecutor.execute(
() ->
ExceptionUtils.rethrow(
new ActionTaskExecutionException(
- "Failed to execute action task", e)),
+ "Failed to execute action task", t)),
"throw exception in mailbox");
}
}
diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java
index 9ac18e3e..22c0072f 100644
--- a/runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java
+++ b/runtime/src/test/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperatorTest.java
@@ -289,6 +289,27 @@ public class ActionExecutionOperatorTest {
}
}
+ @Test
+ void testMailboxSubmittedActionTaskPropagatesError() throws Exception {
+ try (KeyedOneInputStreamOperatorTestHarness<Long, Long, Object> testHarness =
+ new KeyedOneInputStreamOperatorTestHarness<>(
+ new ActionExecutionOperatorFactory(
+ TestAgent.getLinkageErrorAgentPlan(), true),
+ (KeySelector<Long, Long>) value -> value,
+ TypeInformation.of(Long.class))) {
+ testHarness.open();
+ ActionExecutionOperator<Long, Object> operator =
+ (ActionExecutionOperator<Long, Object>) testHarness.getOperator();
+
+ testHarness.processElement(new StreamRecord<>(0L));
+ assertThatThrownBy(() -> operator.waitInFlightEventsFinished())
+ .hasCauseInstanceOf(ActionExecutionOperator.ActionTaskExecutionException.class)
+ .rootCause()
+ .isInstanceOf(NoClassDefFoundError.class)
+ .hasMessageContaining("synthetic missing runtime dependency");
+ }
+ }
+
@Test
void testInMemoryActionStateStoreIntegration() throws Exception {
AgentPlan agentPlanWithStateStore = TestAgent.getAgentPlan(false);
@@ -1711,6 +1732,20 @@ public class ActionExecutionOperatorTest {
}
}
+ public static class LinkageErrorAction {
+ static {
+ if (shouldThrowLinkageError()) {
+ throw new NoClassDefFoundError("synthetic missing runtime dependency");
+ }
+ }
+
+ private static boolean shouldThrowLinkageError() {
+ return true;
+ }
+
+ public static void action(Event event, RunnerContext context) {}
+ }
+
public static void resetReconcilableRecoveryFixture() {
RECONCILABLE_CALL_COUNTER.set(0);
RECONCILABLE_RECONCILE_COUNTER.set(0);
@@ -1937,6 +1972,29 @@ public class ActionExecutionOperatorTest {
return null;
}
+ public static AgentPlan getLinkageErrorAgentPlan() {
+ try {
+ Map<String, List<Action>> actionsByEvent = new HashMap<>();
+ Map<String, Action> actions = new HashMap<>();
+
+ Action errorAction =
+ new Action(
+ "linkageErrorAction",
+ new JavaFunction(
+ LinkageErrorAction.class,
+ "action",
+ new Class<?>[] {Event.class, RunnerContext.class}),
+ Collections.singletonList(InputEvent.EVENT_TYPE));
+ actionsByEvent.put(InputEvent.EVENT_TYPE, Collections.singletonList(errorAction));
+ actions.put(errorAction.getName(), errorAction);
+
+ return new AgentPlan(actions, actionsByEvent, new HashMap<>());
+ } catch (Exception e) {
+ ExceptionUtils.rethrow(e);
+ }
+ return null;
+ }
+
// ==================== Actions for Exception Recovery Tests ====================
/**
(flink-agents) 03/05: [runtime] Binary serde for MemoryUpdate values via versioned Kryo envelope (#874)
Posted by we...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch release-0.3
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
commit 06866c56fd344a7d9ad6a89d6118c265adcb4f8e
Author: Weiqing Yang <ya...@gmail.com>
AuthorDate: Wed Jul 8 02:47:49 2026 -0700
[runtime] Binary serde for MemoryUpdate values via versioned Kryo envelope (#874)
(cherry picked from commit 84c76fbc3e0accb1e4c12c522f28c01b1758793f)
---
.../runtime/actionstate/ActionStateSerde.java | 119 ++++++++++++++++++
.../runtime/actionstate/ActionStateSerdeTest.java | 137 +++++++++++++++++++++
2 files changed, 256 insertions(+)
diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateSerde.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateSerde.java
index 90885e31..d38a4495 100644
--- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateSerde.java
+++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateSerde.java
@@ -19,14 +19,29 @@ package org.apache.flink.agents.runtime.actionstate;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.JsonSerializer;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import com.fasterxml.jackson.databind.module.SimpleModule;
import org.apache.flink.agents.api.Event;
+import org.apache.flink.agents.api.context.MemoryUpdate;
import org.apache.flink.agents.runtime.operator.ActionTask;
+import org.apache.flink.api.common.serialization.SerializerConfigImpl;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.util.Base64;
/**
* Backend-agnostic serializer/deserializer for {@link ActionState}.
@@ -37,6 +52,37 @@ import java.io.IOException;
*/
public final class ActionStateSerde {
+ /**
+ * Recovery-compat contract for the {@link MemoryUpdate#getValue() MemoryUpdate.value} envelope.
+ *
+ * <p>A non-null value is written on disk as {@code
+ * {"serde":"kryo","version":1,"payload":<b64>}} where {@code payload} is the base64 of the
+ * Flink-native (Kryo) binary of the untyped value. That binary is only guaranteed readable by
+ * the same code and same Flink version that wrote it — which is all that ever happens, because
+ * the durable-execution journal is checkpoint-scoped and never survives a code or Flink
+ * upgrade. Bump {@code version} if the payload encoding ever changes; Kryo is pinned within a
+ * Flink major release (supplied by the cluster).
+ */
+ private static final String MEMORY_UPDATE_SERDE = "kryo";
+
+ private static final int MEMORY_UPDATE_VERSION = 1;
+ private static final String ENVELOPE_SERDE_FIELD = "serde";
+ private static final String ENVELOPE_VERSION_FIELD = "version";
+ private static final String ENVELOPE_PAYLOAD_FIELD = "payload";
+
+ // The Flink-native serializer for an untyped Object resolves to Kryo
+ // (GenericTypeInfo<Object> -> KryoSerializer).
+ private static final TypeSerializer<Object> VALUE_SERIALIZER_TEMPLATE =
+ TypeInformation.of(Object.class).createSerializer(new SerializerConfigImpl());
+
+ // KryoSerializer is not thread-safe; only duplicate() (which is safe) is ever called on the
+ // shared template, giving each thread an independent copy for serialize/deserialize. Because
+ // this ThreadLocal is static, a thread's copy can retain (pin) the classes of the values it has
+ // serialized for that thread's lifetime — an accepted tradeoff over rebuilding a heavy Kryo
+ // instance on every call.
+ private static final ThreadLocal<TypeSerializer<Object>> VALUE_SERIALIZER =
+ ThreadLocal.withInitial(VALUE_SERIALIZER_TEMPLATE::duplicate);
+
private static final ObjectMapper OBJECT_MAPPER = createObjectMapper();
private ActionStateSerde() {}
@@ -70,9 +116,25 @@ public final class ActionStateSerde {
module.addSerializer(ActionTask.class, new ActionTaskSerializer());
mapper.registerModule(module);
+ // Preserve the concrete runtime type of the untyped MemoryUpdate.value across durable
+ // recovery by wrapping it in a versioned Kryo envelope.
+ mapper.addMixIn(MemoryUpdate.class, MemoryUpdateValueMixin.class);
+
return mapper;
}
+ private static byte[] kryoSerialize(Object value) throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ VALUE_SERIALIZER.get().serialize(value, new DataOutputViewStreamWrapper(baos));
+ return baos.toByteArray();
+ }
+
+ private static Object kryoDeserialize(byte[] bytes) throws IOException {
+ return VALUE_SERIALIZER
+ .get()
+ .deserialize(new DataInputViewStreamWrapper(new ByteArrayInputStream(bytes)));
+ }
+
/** Mixin to add type information for Event hierarchy. */
@JsonTypeInfo(
use = JsonTypeInfo.Id.CLASS,
@@ -88,4 +150,61 @@ public final class ActionStateSerde {
gen.writeNull();
}
}
+
+ /**
+ * Binds the Kryo envelope (de)serializer to the {@code value} property of {@link MemoryUpdate}.
+ */
+ abstract static class MemoryUpdateValueMixin {
+ @JsonSerialize(using = MemoryUpdateValueSerializer.class)
+ @JsonDeserialize(using = MemoryUpdateValueDeserializer.class)
+ Object value;
+ }
+
+ /**
+ * Writes a non-null {@link MemoryUpdate} value as the {@code {serde,version,payload}} envelope.
+ */
+ static class MemoryUpdateValueSerializer extends JsonSerializer<Object> {
+ @Override
+ public void serialize(Object value, JsonGenerator gen, SerializerProvider provider)
+ throws IOException {
+ gen.writeStartObject();
+ gen.writeStringField(ENVELOPE_SERDE_FIELD, MEMORY_UPDATE_SERDE);
+ gen.writeNumberField(ENVELOPE_VERSION_FIELD, MEMORY_UPDATE_VERSION);
+ gen.writeStringField(
+ ENVELOPE_PAYLOAD_FIELD,
+ Base64.getEncoder().encodeToString(kryoSerialize(value)));
+ gen.writeEndObject();
+ }
+ }
+
+ /** Reads the Kryo envelope back to the concrete value and version-checks it. */
+ static class MemoryUpdateValueDeserializer extends JsonDeserializer<Object> {
+ @Override
+ public Object deserialize(JsonParser p, DeserializationContext ctxt) throws IOException {
+ JsonNode node = ctxt.readTree(p);
+ // Every non-null value this serializer writes is an envelope, and the durable journal
+ // never outlives the code that wrote it (see the recovery-compat contract above), so a
+ // non-envelope node is an unsupported cross-version restore rather than legacy data to
+ // tolerate. Reject it instead of silently binding it to a wrong-typed stock value.
+ if (!isEnvelope(node)) {
+ throw new IOException(
+ "Expected a Kryo-envelope MemoryUpdate value but found: "
+ + node.getNodeType());
+ }
+ int version = node.get(ENVELOPE_VERSION_FIELD).asInt();
+ if (version != MEMORY_UPDATE_VERSION) {
+ throw new IOException(
+ "Unsupported MemoryUpdate value envelope version: " + version);
+ }
+ return kryoDeserialize(
+ Base64.getDecoder().decode(node.get(ENVELOPE_PAYLOAD_FIELD).asText()));
+ }
+
+ private static boolean isEnvelope(JsonNode n) {
+ return n.isObject()
+ && MEMORY_UPDATE_SERDE.equals(n.path(ENVELOPE_SERDE_FIELD).asText())
+ && n.has(ENVELOPE_VERSION_FIELD)
+ && n.path(ENVELOPE_PAYLOAD_FIELD).isTextual();
+ }
+ }
}
diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateSerdeTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateSerdeTest.java
index 116aeeb3..f8dbc6fa 100644
--- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateSerdeTest.java
+++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateSerdeTest.java
@@ -320,6 +320,143 @@ public class ActionStateSerdeTest {
"exception".getBytes(StandardCharsets.UTF_8), legacyFailure.getExceptionPayload());
}
+ @Test
+ public void testByteArrayMemoryValuePreserved() throws Exception {
+ byte[] value = new byte[] {1, 2, 3, 4, 5};
+ ActionState originalState = new ActionState(new InputEvent("in"));
+ originalState.addShortTermMemoryUpdate(new MemoryUpdate("stm.bytes", value));
+
+ ActionState deserializedState =
+ ActionStateSerde.deserialize(ActionStateSerde.serialize(originalState));
+
+ Object recovered = deserializedState.getShortTermMemoryUpdates().get(0).getValue();
+ assertInstanceOf(byte[].class, recovered);
+ assertArrayEquals(value, (byte[]) recovered);
+ }
+
+ @Test
+ public void testLongMemoryValuePreserved() throws Exception {
+ Long value = 42L;
+ ActionState originalState = new ActionState(new InputEvent("in"));
+ originalState.addShortTermMemoryUpdate(new MemoryUpdate("stm.long", value));
+
+ ActionState deserializedState =
+ ActionStateSerde.deserialize(ActionStateSerde.serialize(originalState));
+
+ Object recovered = deserializedState.getShortTermMemoryUpdates().get(0).getValue();
+ assertInstanceOf(Long.class, recovered);
+ assertEquals(42L, recovered);
+ }
+
+ @Test
+ public void testNestedCollectionMemoryValuePreserved() throws Exception {
+ byte[] nestedBytes = new byte[] {9, 8, 7};
+ List<Object> innerList = new ArrayList<>();
+ innerList.add(nestedBytes);
+ innerList.add(7L);
+ Map<String, Object> value = new HashMap<>();
+ value.put("items", innerList);
+
+ ActionState originalState = new ActionState(new InputEvent("in"));
+ originalState.addShortTermMemoryUpdate(new MemoryUpdate("stm.nested", value));
+
+ ActionState deserializedState =
+ ActionStateSerde.deserialize(ActionStateSerde.serialize(originalState));
+
+ Object recovered = deserializedState.getShortTermMemoryUpdates().get(0).getValue();
+ assertInstanceOf(Map.class, recovered);
+ @SuppressWarnings("unchecked")
+ List<Object> recoveredList = (List<Object>) ((Map<String, Object>) recovered).get("items");
+ assertInstanceOf(byte[].class, recoveredList.get(0));
+ assertArrayEquals(nestedBytes, (byte[]) recoveredList.get(0));
+ assertInstanceOf(Long.class, recoveredList.get(1));
+ assertEquals(7L, recoveredList.get(1));
+ }
+
+ @Test
+ public void testPojoMemoryValuePreserved() throws Exception {
+ MemoryValuePojo value = new MemoryValuePojo("hello", 99);
+ ActionState originalState = new ActionState(new InputEvent("in"));
+ originalState.addShortTermMemoryUpdate(new MemoryUpdate("stm.pojo", value));
+
+ ActionState deserializedState =
+ ActionStateSerde.deserialize(ActionStateSerde.serialize(originalState));
+
+ Object recovered = deserializedState.getShortTermMemoryUpdates().get(0).getValue();
+ assertInstanceOf(MemoryValuePojo.class, recovered);
+ assertEquals(value, recovered);
+ }
+
+ @Test
+ public void testNullMemoryValuePreserved() throws Exception {
+ ActionState originalState = new ActionState(new InputEvent("in"));
+ originalState.addShortTermMemoryUpdate(new MemoryUpdate("stm.null", null));
+
+ ActionState deserializedState =
+ ActionStateSerde.deserialize(ActionStateSerde.serialize(originalState));
+
+ assertEquals(1, deserializedState.getShortTermMemoryUpdates().size());
+ assertNull(deserializedState.getShortTermMemoryUpdates().get(0).getValue());
+ }
+
+ @Test
+ public void testNonEnvelopeMemoryValueRejected() throws Exception {
+ ActionState originalState = new ActionState(new InputEvent("in"));
+ originalState.addShortTermMemoryUpdate(new MemoryUpdate("stm.raw", "some value"));
+
+ byte[] serialized = ActionStateSerde.serialize(originalState);
+ String json = new String(serialized, StandardCharsets.UTF_8);
+ // Replace the whole envelope object with a bare JSON value, as a legacy pre-envelope
+ // journal would have stored it.
+ int start = json.indexOf("{\"serde\":");
+ int end = json.indexOf('}', start) + 1;
+ byte[] patched =
+ (json.substring(0, start) + "\"some value\"" + json.substring(end))
+ .getBytes(StandardCharsets.UTF_8);
+
+ assertThrows(RuntimeException.class, () -> ActionStateSerde.deserialize(patched));
+ }
+
+ @Test
+ public void testUnknownEnvelopeVersionRejected() throws Exception {
+ ActionState originalState = new ActionState(new InputEvent("in"));
+ originalState.addShortTermMemoryUpdate(new MemoryUpdate("stm.version", "some value"));
+
+ byte[] serialized = ActionStateSerde.serialize(originalState);
+ String json = new String(serialized, StandardCharsets.UTF_8);
+ assertTrue(json.contains("\"version\":1"));
+ byte[] patched =
+ json.replace("\"version\":1", "\"version\":2").getBytes(StandardCharsets.UTF_8);
+
+ assertThrows(RuntimeException.class, () -> ActionStateSerde.deserialize(patched));
+ }
+
+ /** Serializable POJO used to verify Kryo preserves user types across durable recovery. */
+ public static class MemoryValuePojo implements java.io.Serializable {
+ private String name;
+ private int count;
+
+ public MemoryValuePojo() {}
+
+ public MemoryValuePojo(String name, int count) {
+ this.name = name;
+ this.count = count;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (!(o instanceof MemoryValuePojo)) return false;
+ MemoryValuePojo that = (MemoryValuePojo) o;
+ return count == that.count && java.util.Objects.equals(name, that.name);
+ }
+
+ @Override
+ public int hashCode() {
+ return java.util.Objects.hash(name, count);
+ }
+ }
+
@Test
public void testKafkaSederDelegatesToActionStateSerde() throws Exception {
InputEvent inputEvent = new InputEvent("test delegation");
(flink-agents) 04/05: [tools] Update install.sh for Flink Agents 0.3.0 and Python 3.12 (#888)
Posted by we...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch release-0.3
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
commit 77aedb9c31dc978aa879109963f6398b81111d8b
Author: Femi <44...@users.noreply.github.com>
AuthorDate: Tue Jul 14 04:46:27 2026 +0100
[tools] Update install.sh for Flink Agents 0.3.0 and Python 3.12 (#888)
(cherry picked from commit 96b943edf016c96f2394ad58eb10efeb7251372a)
---
tools/install.sh | 133 ++++++++++++++--
tools/test/helpers/load.bash | 7 +-
tools/test/integration/dry_run_extra.bats | 2 +-
tools/test/unit/revalidate_python_constraint.bats | 175 ++++++++++++++++++++++
tools/test/unit/validate_python_bin.bats | 39 ++++-
5 files changed, 341 insertions(+), 15 deletions(-)
diff --git a/tools/install.sh b/tools/install.sh
index 4fcd3db0..70fc516f 100755
--- a/tools/install.sh
+++ b/tools/install.sh
@@ -445,7 +445,7 @@ mark_explicit INSTALL_DIR
mark_explicit VENV_DIR
FLINK_VERSION="${FLINK_VERSION:-2.2.1}"
-FLINK_AGENTS_VERSION="${FLINK_AGENTS_VERSION:-0.2.1}"
+FLINK_AGENTS_VERSION="${FLINK_AGENTS_VERSION:-0.3.0}"
FLINK_SCALA_VERSION="${FLINK_SCALA_VERSION:-2.12}"
FLINK_BASE_URL="${FLINK_BASE_URL:-https://dlcdn.apache.org/flink}"
# Flink Agents JARs live next to Flink on the ASF mirror network. Override
@@ -460,10 +460,10 @@ FLINK_RECOMMENDED_VERSION="2.2.1"
# Mirrors https://flink.apache.org/downloads/#apache-flink-agents
# (latest first). Note: 0.1.x only ships JARs for Flink 1.20, while 0.2.x
-# ships JARs for Flink 1.20 / 2.0 / 2.1 / 2.2 — see the download page for
-# the exact compatibility matrix.
-FLINK_AGENTS_SUPPORTED_VERSIONS=("0.2.1" "0.2.0" "0.1.1" "0.1.0")
-FLINK_AGENTS_RECOMMENDED_VERSION="0.2.1"
+# and 0.3.x ship JARs for Flink 1.20 / 2.0 / 2.1 / 2.2 — see the download
+# page for the exact compatibility matrix.
+FLINK_AGENTS_SUPPORTED_VERSIONS=("0.3.0" "0.2.1" "0.2.0" "0.1.1" "0.1.0")
+FLINK_AGENTS_RECOMMENDED_VERSION="0.3.0"
INSTALL_FLINK="${INSTALL_FLINK:-Ask}"
ENABLE_PYFLINK="${ENABLE_PYFLINK:-Ask}"
@@ -475,6 +475,9 @@ VERBOSE="${FLINK_AGENTS_VERBOSE:-0}"
DRY_RUN="${FLINK_AGENTS_DRY_RUN:-0}"
HELP=0
PYFLINK_ACTUALLY_ENABLED=0
+# Set to 1 when a version edit leaves an existing venv on an incompatible
+# interpreter and the user opts to recreate it; setup_python_env honors it.
+RECREATE_VENV=0
print_usage() {
cat <<EOF
@@ -1015,6 +1018,7 @@ edit_plan_dump_state() {
printf 'VENV_DIR=%s\n' "$(edit_plan_quote "$VENV_DIR")"
printf 'PYTHON_BIN=%s\n' "$(edit_plan_quote "${PYTHON_BIN:-}")"
printf 'FLINK_AGENTS_VERSION=%s\n' "$(edit_plan_quote "$FLINK_AGENTS_VERSION")"
+ printf 'RECREATE_VENV=%s\n' "$(edit_plan_quote "${RECREATE_VENV:-0}")"
} > "$out"
}
@@ -1111,6 +1115,7 @@ edit_plan_interactive() {
case "$action" in
flink_agents_version)
prompt_flink_agents_version_interactive || true
+ revalidate_python_constraint
;;
install_flink)
if choose_install_method_interactive "Install Flink?"; then
@@ -1133,11 +1138,13 @@ edit_plan_interactive() {
detect_flink_version_from_home || ui_warn "Could not auto-detect Flink version; keeping ${FLINK_VERSION}"
fi
FLINK_MAJOR_MINOR="$(flink_major_minor "$FLINK_VERSION")"
+ revalidate_python_constraint
;;
flink_version)
prompt_flink_version_interactive || true
FLINK_HOME="${INSTALL_DIR}/flink-${FLINK_VERSION}"
FLINK_MAJOR_MINOR="$(flink_major_minor "$FLINK_VERSION")"
+ revalidate_python_constraint
;;
install_dir)
INSTALL_DIR="$(prompt_path_choice_interactive \
@@ -1404,10 +1411,25 @@ pip_install_quiet() {
}
setup_python_env() {
+ # A venv the user opted to recreate (incompatible interpreter, see
+ # revalidate_existing_venv) is deleted here, not at plan time, so a later
+ # cancel never destroys it. Guard the rm to a real venv marker.
+ if [[ "${RECREATE_VENV:-0}" == "1" && -f "$VENV_DIR/pyvenv.cfg" ]]; then
+ ui_info "Recreating virtual environment: $VENV_DIR"
+ rm -rf "$VENV_DIR"
+ fi
+
if [[ ! -d "$VENV_DIR" ]]; then
ui_info "Creating virtual environment: $VENV_DIR"
create_venv
else
+ # Reused venvs keep their own interpreter; make sure it still fits the
+ # selected versions before pip runs under it (belt-and-suspenders for
+ # any path that bypassed the plan-time revalidation, e.g. an explicit
+ # VENV_DIR in non-interactive mode).
+ if [[ -x "$VENV_DIR/bin/python" ]] && ! validate_python_bin "$VENV_DIR/bin/python"; then
+ die "Existing venv ${VENV_DIR} uses an interpreter incompatible with Flink Agents ${FLINK_AGENTS_VERSION} + Flink ${FLINK_VERSION} (needs >=3.10 and <3.$(python_minor_ceiling)). Remove it or set VENV_DIR to a different path."
+ fi
ui_info "Reusing existing virtual environment: $VENV_DIR"
fi
@@ -1563,6 +1585,38 @@ check_java() {
return 0
}
+# Upper bound (exclusive) on the Python minor version, derived from BOTH
+# selected versions and returning the stricter (lower) of the two:
+# - Flink Agents: 0.1.x / 0.2.x publish requires-python >=3.10,<3.12,
+# while 0.3.0+ publishes >=3.10,<3.13 (adds Python 3.12 support).
+# - Apache Flink: Python 3.12 requires Flink 2.1+; Flink <2.1 caps at <3.12.
+# An unparseable FLINK_VERSION leaves the Flink axis unrestricted (fail open),
+# so only the Flink Agents ceiling applies in that case.
+python_minor_ceiling() {
+ local agents_ceiling flink_ceiling
+ case "$FLINK_AGENTS_VERSION" in
+ 0.1.*|0.2.*) agents_ceiling=12 ;;
+ *) agents_ceiling=13 ;;
+ esac
+
+ flink_ceiling=13
+ local mm major minor
+ mm="$(flink_major_minor "${FLINK_VERSION:-}")"
+ if [[ -n "$mm" ]]; then
+ major="${mm%%.*}"
+ minor="${mm##*.}"
+ if (( major < 2 || (major == 2 && minor < 1) )); then
+ flink_ceiling=12
+ fi
+ fi
+
+ if (( agents_ceiling < flink_ceiling )); then
+ printf '%s' "$agents_ceiling"
+ else
+ printf '%s' "$flink_ceiling"
+ fi
+}
+
validate_python_bin() {
local bin="$1"
[[ -n "$bin" ]] || return 1
@@ -1576,7 +1630,7 @@ validate_python_bin() {
py_major="${py_version_output%%.*}"
py_minor="${py_version_output##*.}"
- if [[ "$py_major" -ne 3 ]] || [[ "$py_minor" -lt 10 ]] || [[ "$py_minor" -ge 12 ]]; then
+ if [[ "$py_major" -ne 3 ]] || [[ "$py_minor" -lt 10 ]] || [[ "$py_minor" -ge "$(python_minor_ceiling)" ]]; then
return 1
fi
return 0
@@ -1588,7 +1642,7 @@ resolve_python() {
ui_success "Using Python: $PYTHON_BIN"
return 0
fi
- die "PYTHON_BIN is invalid or unsupported (need >=3.10 and <3.12): $PYTHON_BIN"
+ die "PYTHON_BIN is invalid or unsupported (need >=3.10 and <3.$(python_minor_ceiling)): $PYTHON_BIN"
fi
if validate_python_bin python3; then
@@ -1600,7 +1654,7 @@ resolve_python() {
fi
if command -v python3 >/dev/null 2>&1; then
- ui_warn "python3 on PATH is incompatible (Flink Agents requires Python >=3.10 and <3.12)"
+ ui_warn "python3 on PATH is incompatible (Flink Agents ${FLINK_AGENTS_VERSION} requires Python >=3.10 and <3.$(python_minor_ceiling))"
else
ui_warn "python3 not found on PATH"
fi
@@ -1615,13 +1669,74 @@ resolve_python() {
die "No Python interpreter provided."
fi
if ! validate_python_bin "$input"; then
- die "Provided Python is invalid or unsupported (need >=3.10 and <3.12): $input"
+ die "Provided Python is invalid or unsupported (need >=3.10 and <3.$(python_minor_ceiling)): $input"
fi
PYTHON_BIN="$input"
ui_success "Using Python: $PYTHON_BIN"
return 0
}
+# The Python ceiling depends on BOTH the Flink Agents and the Flink version
+# (see python_minor_ceiling), so editing either at the confirm screen can
+# invalidate a Python setup that was already accepted — e.g. Python 3.12
+# passes for 0.3.0 + Flink 2.2 but not after editing down to 0.2.1 or to
+# Flink 2.0. Called after a version edit; mirrors the enable_pyflink arm.
+# No-op unless PyFlink is enabled. Two independent checks:
+# 1. PYTHON_BIN — the resolved interpreter; re-resolve if it no longer fits.
+# 2. VENV_DIR — see revalidate_existing_venv; a reused venv keeps its own
+# interpreter, which re-resolving PYTHON_BIN does not change.
+revalidate_python_constraint() {
+ if [[ "$ENABLE_PYFLINK" != "Yes" ]]; then
+ return 0
+ fi
+
+ if [[ -n "${PYTHON_BIN:-}" ]] && ! validate_python_bin "$PYTHON_BIN"; then
+ ui_warn "The selected Python (${PYTHON_BIN}) is incompatible with Flink Agents ${FLINK_AGENTS_VERSION} + Flink ${FLINK_VERSION} (requires Python >=3.10 and <3.$(python_minor_ceiling))"
+ PYTHON_BIN=""
+ resolve_python
+ fi
+
+ revalidate_existing_venv
+}
+
+# setup_python_env reuses an existing venv as-is (it only creates one when
+# VENV_DIR is absent), so the venv's own interpreter — not the re-resolved
+# PYTHON_BIN — is what pip runs under. After a version edit that lowers the
+# Python ceiling, an existing venv built on a now-too-new interpreter would
+# still be reused and fail at `pip install` time. Catch it at plan time:
+# offer to recreate it (deferred to setup_python_env via RECREATE_VENV, which
+# only ever deletes a directory carrying a pyvenv.cfg marker) or point at a
+# different path. Non-interactive callers get an actionable error instead of a
+# mid-install failure. No-op unless VENV_DIR is an existing venv.
+revalidate_existing_venv() {
+ RECREATE_VENV=0
+ [[ "$ENABLE_PYFLINK" == "Yes" && -n "${VENV_DIR:-}" ]] || return 0
+
+ local kind
+ kind="$(validate_venv_dir "$VENV_DIR")" || true
+ [[ "$kind" == "venv" ]] || return 0
+
+ if validate_python_bin "$VENV_DIR/bin/python"; then
+ return 0
+ fi
+
+ local venv_pv
+ venv_pv="$("$VENV_DIR/bin/python" -c 'import sys; print(f"{sys.version_info.major}.{sys.version_info.minor}")' 2>/dev/null || printf 'unknown')"
+ ui_warn "Existing venv ${VENV_DIR} uses Python ${venv_pv}, incompatible with Flink Agents ${FLINK_AGENTS_VERSION} + Flink ${FLINK_VERSION} (needs >=3.10 and <3.$(python_minor_ceiling))"
+
+ if ! is_promptable; then
+ die "Recreate ${VENV_DIR} or set VENV_DIR to a different path (its interpreter ${venv_pv} is unsupported for this version selection)."
+ fi
+
+ if choose_install_method_interactive "Recreate ${VENV_DIR} with a compatible interpreter?"; then
+ RECREATE_VENV=1
+ else
+ prompt_and_validate_venv_dir "$VENV_DIR"
+ # The newly chosen path may itself be an incompatible venv.
+ revalidate_existing_venv
+ fi
+}
+
show_install_plan() {
ui_section "Environment (read-only)"
ui_kv "OS" "$OS"
diff --git a/tools/test/helpers/load.bash b/tools/test/helpers/load.bash
index 043a4fd6..e2cae738 100644
--- a/tools/test/helpers/load.bash
+++ b/tools/test/helpers/load.bash
@@ -28,8 +28,9 @@ reset_install_sh_state() {
ENABLE_PYFLINK="Ask"
PYTHON_BIN=""
PYFLINK_ACTUALLY_ENABLED=0
+ RECREATE_VENV=0
FLINK_VERSION="2.2.0"
- FLINK_AGENTS_VERSION="0.2.1"
+ FLINK_AGENTS_VERSION="0.3.0"
FLINK_SCALA_VERSION="2.12"
FLINK_BASE_URL="https://dlcdn.apache.org/flink"
FLINK_SUPPORTED_VERSIONS=("2.2.0" "2.1.1" "2.0.1" "1.20.3")
@@ -44,6 +45,6 @@ reset_install_sh_state() {
FLINK_AGENTS_VERSION_EXPLICIT=0
INSTALL_DIR_EXPLICIT=0
VENV_DIR_EXPLICIT=0
- FLINK_AGENTS_SUPPORTED_VERSIONS=("0.2.1" "0.2.0" "0.1.1" "0.1.0")
- FLINK_AGENTS_RECOMMENDED_VERSION="0.2.1"
+ FLINK_AGENTS_SUPPORTED_VERSIONS=("0.3.0" "0.2.1" "0.2.0" "0.1.1" "0.1.0")
+ FLINK_AGENTS_RECOMMENDED_VERSION="0.3.0"
}
diff --git a/tools/test/integration/dry_run_extra.bats b/tools/test/integration/dry_run_extra.bats
index 1e7efac0..105ccf4c 100644
--- a/tools/test/integration/dry_run_extra.bats
+++ b/tools/test/integration/dry_run_extra.bats
@@ -19,7 +19,7 @@ esac
@test "dry-run: plan shows Flink Agents version (review #1 — JARs are implicit)" {
run bash "${BATS_TEST_DIRNAME}/../../install.sh" --dry-run --non-interactive
[ "$status" -eq 0 ]
- case "$output" in *"Flink Agents version"*"0.2.1"*) ;; *) false ;; esac
+ case "$output" in *"Flink Agents version"*"0.3.0"*) ;; *) false ;; esac
}
@test "dry-run: existing FLINK_HOME — plan shows detected version, not default (review #2)" {
diff --git a/tools/test/unit/revalidate_python_constraint.bats b/tools/test/unit/revalidate_python_constraint.bats
new file mode 100644
index 00000000..354df87a
--- /dev/null
+++ b/tools/test/unit/revalidate_python_constraint.bats
@@ -0,0 +1,175 @@
+#!/usr/bin/env bats
+
+setup() {
+ load '../helpers/load'
+ load '../helpers/shim'
+ load_install_sh
+ reset_install_sh_state
+ shim_setup
+ # Default the venv to an absent path so revalidate_existing_venv is a
+ # no-op unless a test builds a real venv on purpose.
+ VENV_DIR="$BATS_TEST_TMPDIR/no-venv"
+}
+
+# Helper: write a fake python that reports a chosen version.
+fake_python() {
+ local name="$1"
+ local ver="$2"
+ shim_bin_script "$name" "
+case \"\$*\" in
+ *'sys.version_info.major'*)
+ echo '$ver'
+ ;;
+esac
+"
+}
+
+# Helper: build a fake existing venv whose interpreter reports $ver.
+fake_venv() {
+ local dir="$1"
+ local ver="$2"
+ mkdir -p "$dir/bin"
+ : > "$dir/pyvenv.cfg"
+ : > "$dir/bin/activate"
+ cat > "$dir/bin/python" <<EOF
+#!/usr/bin/env bash
+case "\$*" in
+ *'sys.version_info.major'*) echo '$ver' ;;
+esac
+EOF
+ chmod +x "$dir/bin/python"
+}
+
+# --- PYTHON_BIN re-resolution (Flink Agents axis) ---
+
+# Scenario from review: interpreter resolved under 0.3.0 (ceiling 3.13),
+# then the version is edited down to 0.2.1 (ceiling 3.12) at the confirm
+# screen. The stale interpreter must be dropped and re-resolved.
+@test "revalidate_python_constraint: re-resolves when the edited Agents version drops the ceiling below PYTHON_BIN" {
+ fake_python fake_py312 "3.12"
+ fake_python python3 "3.11"
+
+ ENABLE_PYFLINK="Yes"
+ PYTHON_BIN="fake_py312"
+ FLINK_AGENTS_VERSION="0.2.1"
+
+ revalidate_python_constraint
+ [ "$PYTHON_BIN" = "python3" ]
+}
+
+@test "revalidate_python_constraint: keeps an interpreter that still satisfies the new version" {
+ fake_python fake_py312 "3.12"
+
+ ENABLE_PYFLINK="Yes"
+ PYTHON_BIN="fake_py312"
+ FLINK_AGENTS_VERSION="0.3.0"
+
+ revalidate_python_constraint
+ [ "$PYTHON_BIN" = "fake_py312" ]
+}
+
+@test "revalidate_python_constraint: no-op when PyFlink is disabled" {
+ fake_python fake_py312 "3.12"
+
+ ENABLE_PYFLINK="No"
+ PYTHON_BIN="fake_py312"
+ FLINK_AGENTS_VERSION="0.2.1"
+
+ revalidate_python_constraint
+ [ "$PYTHON_BIN" = "fake_py312" ]
+}
+
+@test "revalidate_python_constraint: no-op when no interpreter was resolved yet" {
+ ENABLE_PYFLINK="Yes"
+ PYTHON_BIN=""
+ FLINK_AGENTS_VERSION="0.2.1"
+
+ run revalidate_python_constraint
+ [ "$status" -eq 0 ]
+}
+
+@test "revalidate_python_constraint: warns before re-resolving a stale interpreter" {
+ fake_python fake_py312 "3.12"
+ fake_python python3 "3.11"
+
+ ENABLE_PYFLINK="Yes"
+ PYTHON_BIN="fake_py312"
+ FLINK_AGENTS_VERSION="0.2.1"
+
+ run revalidate_python_constraint
+ [ "$status" -eq 0 ]
+ [[ "$output" == *"incompatible with Flink Agents 0.2.1"* ]]
+ [[ "$output" == *"<3.12"* ]]
+}
+
+# --- PYTHON_BIN re-resolution (Flink axis) ---
+
+# Python 3.12 needs Flink 2.1+. Editing the Flink version down to 2.0.x while
+# Agents stays 0.3.0 must still drop a 3.12 interpreter.
+@test "revalidate_python_constraint: re-resolves when the edited Flink version drops the ceiling below PYTHON_BIN" {
+ fake_python fake_py312 "3.12"
+ fake_python python3 "3.11"
+
+ ENABLE_PYFLINK="Yes"
+ PYTHON_BIN="fake_py312"
+ FLINK_AGENTS_VERSION="0.3.0"
+ FLINK_VERSION="2.0.2"
+
+ revalidate_python_constraint
+ [ "$PYTHON_BIN" = "python3" ]
+}
+
+@test "revalidate_python_constraint: keeps a 3.12 interpreter when Flink is 2.1+" {
+ fake_python fake_py312 "3.12"
+
+ ENABLE_PYFLINK="Yes"
+ PYTHON_BIN="fake_py312"
+ FLINK_AGENTS_VERSION="0.3.0"
+ FLINK_VERSION="2.1.3"
+
+ revalidate_python_constraint
+ [ "$PYTHON_BIN" = "fake_py312" ]
+}
+
+# --- existing-venv revalidation ---
+
+# The reviewer's second scenario: PYTHON_BIN re-resolves fine (3.11 is valid
+# for both versions), but an existing venv built on 3.12 would still be reused
+# by setup_python_env. Non-interactive callers must get an actionable error.
+@test "revalidate_existing_venv: non-interactive die when the existing venv interpreter is too new" {
+ fake_python python3 "3.11"
+ fake_venv "$BATS_TEST_TMPDIR/venv312" "3.12"
+
+ NO_PROMPT=1
+ ENABLE_PYFLINK="Yes"
+ PYTHON_BIN="python3"
+ FLINK_AGENTS_VERSION="0.2.1"
+ VENV_DIR="$BATS_TEST_TMPDIR/venv312"
+
+ run revalidate_existing_venv
+ [ "$status" -ne 0 ]
+ [[ "$output" == *"uses Python 3.12"* ]]
+ [[ "$output" == *"different path"* ]]
+}
+
+@test "revalidate_existing_venv: no-op when the existing venv interpreter is compatible" {
+ fake_venv "$BATS_TEST_TMPDIR/venv311" "3.11"
+
+ ENABLE_PYFLINK="Yes"
+ FLINK_AGENTS_VERSION="0.2.1"
+ VENV_DIR="$BATS_TEST_TMPDIR/venv311"
+ RECREATE_VENV=0
+
+ run revalidate_existing_venv
+ [ "$status" -eq 0 ]
+ [ "$RECREATE_VENV" = "0" ]
+}
+
+@test "revalidate_existing_venv: no-op when VENV_DIR is not an existing venv" {
+ ENABLE_PYFLINK="Yes"
+ FLINK_AGENTS_VERSION="0.2.1"
+ VENV_DIR="$BATS_TEST_TMPDIR/does-not-exist"
+
+ run revalidate_existing_venv
+ [ "$status" -eq 0 ]
+}
diff --git a/tools/test/unit/validate_python_bin.bats b/tools/test/unit/validate_python_bin.bats
index 87998bd1..2b99ecda 100644
--- a/tools/test/unit/validate_python_bin.bats
+++ b/tools/test/unit/validate_python_bin.bats
@@ -48,18 +48,53 @@ esac
[ "$status" -ne 0 ]
}
-@test "validate_python_bin: Python 3.12 is rejected" {
+@test "validate_python_bin: Python 3.12 is rejected for Flink Agents 0.2.x" {
+ FLINK_AGENTS_VERSION="0.2.1"
fake_python "3.12"
run validate_python_bin fake_py
[ "$status" -ne 0 ]
}
-@test "validate_python_bin: Python 3.13 is rejected" {
+@test "validate_python_bin: Python 3.12 is accepted for Flink Agents 0.3.0+" {
+ FLINK_AGENTS_VERSION="0.3.0"
+ fake_python "3.12"
+ run validate_python_bin fake_py
+ [ "$status" -eq 0 ]
+}
+
+@test "validate_python_bin: Python 3.13 is rejected for Flink Agents 0.3.0+" {
+ FLINK_AGENTS_VERSION="0.3.0"
fake_python "3.13"
run validate_python_bin fake_py
[ "$status" -ne 0 ]
}
+# Python 3.12 support also requires Flink 2.1+, independent of the Agents
+# version: 0.3.0 alone is not enough if Flink is 2.0.x or 1.20.x.
+@test "validate_python_bin: Python 3.12 is rejected for Flink Agents 0.3.0 on Flink 2.0.x" {
+ FLINK_AGENTS_VERSION="0.3.0"
+ FLINK_VERSION="2.0.2"
+ fake_python "3.12"
+ run validate_python_bin fake_py
+ [ "$status" -ne 0 ]
+}
+
+@test "validate_python_bin: Python 3.12 is rejected for Flink Agents 0.3.0 on Flink 1.20.x" {
+ FLINK_AGENTS_VERSION="0.3.0"
+ FLINK_VERSION="1.20.5"
+ fake_python "3.12"
+ run validate_python_bin fake_py
+ [ "$status" -ne 0 ]
+}
+
+@test "validate_python_bin: Python 3.12 is accepted for Flink Agents 0.3.0 on Flink 2.1.x" {
+ FLINK_AGENTS_VERSION="0.3.0"
+ FLINK_VERSION="2.1.3"
+ fake_python "3.12"
+ run validate_python_bin fake_py
+ [ "$status" -eq 0 ]
+}
+
@test "validate_python_bin: Python 2.7 is rejected" {
fake_python "2.7"
run validate_python_bin fake_py
(flink-agents) 05/05: [tools] Keep install plan edits consistent (#899)
Posted by we...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch release-0.3
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
commit ea2458107c5936a254cb5322796d4bdc1863b7b1
Author: Wenjin Xie <16...@users.noreply.github.com>
AuthorDate: Tue Jul 14 17:48:32 2026 +0800
[tools] Keep install plan edits consistent (#899)
Co-authored-by: Codex <co...@openai.com>
(cherry picked from commit 6f020c5061bc1b444fbdb832d57ee7430517a62a)
---
tools/install.sh | 87 ++++++++++++++++-------
tools/test/helpers/load.bash | 2 +-
tools/test/unit/edit_plan_quote.bats | 3 +
tools/test/unit/reconcile_plan_after_edit.bats | 55 ++++++++++++++
tools/test/unit/revalidate_python_constraint.bats | 73 ++++++++++++++++++-
5 files changed, 190 insertions(+), 30 deletions(-)
diff --git a/tools/install.sh b/tools/install.sh
index 70fc516f..68963ca3 100755
--- a/tools/install.sh
+++ b/tools/install.sh
@@ -475,9 +475,10 @@ VERBOSE="${FLINK_AGENTS_VERBOSE:-0}"
DRY_RUN="${FLINK_AGENTS_DRY_RUN:-0}"
HELP=0
PYFLINK_ACTUALLY_ENABLED=0
-# Set to 1 when a version edit leaves an existing venv on an incompatible
-# interpreter and the user opts to recreate it; setup_python_env honors it.
-RECREATE_VENV=0
+# Exact venv path the user approved for recreation after an edit made its
+# interpreter incompatible. Keeping the path, rather than a boolean, prevents
+# a later VENV_DIR edit from transferring that destructive intent elsewhere.
+RECREATE_VENV_PATH=""
print_usage() {
cat <<EOF
@@ -992,6 +993,29 @@ edit_prompt_flink_home() {
export FLINK_HOME
}
+# Recompute derived values and constraints once after every successful edit.
+# Flink detection is limited to Flink-related actions because invoking
+# bin/flink may start a JVM; Python and venv validation is cheap and applies
+# after every action so path-bound recreation state cannot become stale.
+reconcile_plan_after_edit() {
+ local action="$1"
+ case "$action" in
+ install_flink|flink_version|install_dir|flink_home)
+ if [[ "$INSTALL_FLINK" == "Yes" ]]; then
+ FLINK_HOME="${INSTALL_DIR}/flink-${FLINK_VERSION}"
+ else
+ if detect_flink_version_from_home; then
+ ui_success "Detected Flink version: $FLINK_VERSION"
+ else
+ ui_warn "Could not auto-detect Flink version; keeping ${FLINK_VERSION}"
+ fi
+ fi
+ FLINK_MAJOR_MINOR="$(flink_major_minor "$FLINK_VERSION")"
+ ;;
+ esac
+ revalidate_python_constraint
+}
+
# Show a menu of currently-applicable plan fields and re-prompt for the
# selected one. After the edit, recomputes FLINK_HOME / FLINK_MAJOR_MINOR
# so the next show_install_plan reflects the change.
@@ -1018,7 +1042,7 @@ edit_plan_dump_state() {
printf 'VENV_DIR=%s\n' "$(edit_plan_quote "$VENV_DIR")"
printf 'PYTHON_BIN=%s\n' "$(edit_plan_quote "${PYTHON_BIN:-}")"
printf 'FLINK_AGENTS_VERSION=%s\n' "$(edit_plan_quote "$FLINK_AGENTS_VERSION")"
- printf 'RECREATE_VENV=%s\n' "$(edit_plan_quote "${RECREATE_VENV:-0}")"
+ printf 'RECREATE_VENV_PATH=%s\n' "$(edit_plan_quote "${RECREATE_VENV_PATH:-}")"
} > "$out"
}
@@ -1115,7 +1139,6 @@ edit_plan_interactive() {
case "$action" in
flink_agents_version)
prompt_flink_agents_version_interactive || true
- revalidate_python_constraint
;;
install_flink)
if choose_install_method_interactive "Install Flink?"; then
@@ -1131,20 +1154,13 @@ edit_plan_interactive() {
"/path/to/flink-install-dir")"
INSTALL_DIR="$(normalize_path "$INSTALL_DIR")"
fi
- FLINK_HOME="${INSTALL_DIR}/flink-${FLINK_VERSION}"
else
INSTALL_FLINK=No
edit_prompt_flink_home
- detect_flink_version_from_home || ui_warn "Could not auto-detect Flink version; keeping ${FLINK_VERSION}"
fi
- FLINK_MAJOR_MINOR="$(flink_major_minor "$FLINK_VERSION")"
- revalidate_python_constraint
;;
flink_version)
prompt_flink_version_interactive || true
- FLINK_HOME="${INSTALL_DIR}/flink-${FLINK_VERSION}"
- FLINK_MAJOR_MINOR="$(flink_major_minor "$FLINK_VERSION")"
- revalidate_python_constraint
;;
install_dir)
INSTALL_DIR="$(prompt_path_choice_interactive \
@@ -1152,7 +1168,6 @@ edit_plan_interactive() {
"$INSTALL_DIR" \
"/path/to/flink-install-dir")"
INSTALL_DIR="$(normalize_path "$INSTALL_DIR")"
- FLINK_HOME="${INSTALL_DIR}/flink-${FLINK_VERSION}"
;;
flink_home)
edit_prompt_flink_home
@@ -1174,6 +1189,8 @@ edit_plan_interactive() {
;;
esac
+ reconcile_plan_after_edit "$action"
+
edit_plan_dump_state "$state_file"
) || rc=$?
@@ -1411,12 +1428,15 @@ pip_install_quiet() {
}
setup_python_env() {
- # A venv the user opted to recreate (incompatible interpreter, see
- # revalidate_existing_venv) is deleted here, not at plan time, so a later
- # cancel never destroys it. Guard the rm to a real venv marker.
- if [[ "${RECREATE_VENV:-0}" == "1" && -f "$VENV_DIR/pyvenv.cfg" ]]; then
+ # Recreation is deferred until install time so cancelling the plan never
+ # destroys data. Require both the exact approved path and a real venv
+ # marker before removing anything.
+ if [[ -n "${RECREATE_VENV_PATH:-}" \
+ && "$VENV_DIR" == "$RECREATE_VENV_PATH" \
+ && -f "$VENV_DIR/pyvenv.cfg" ]]; then
ui_info "Recreating virtual environment: $VENV_DIR"
rm -rf "$VENV_DIR"
+ RECREATE_VENV_PATH=""
fi
if [[ ! -d "$VENV_DIR" ]]; then
@@ -1680,13 +1700,14 @@ resolve_python() {
# (see python_minor_ceiling), so editing either at the confirm screen can
# invalidate a Python setup that was already accepted — e.g. Python 3.12
# passes for 0.3.0 + Flink 2.2 but not after editing down to 0.2.1 or to
-# Flink 2.0. Called after a version edit; mirrors the enable_pyflink arm.
+# Flink 2.0. Called by reconcile_plan_after_edit after every menu edit.
# No-op unless PyFlink is enabled. Two independent checks:
# 1. PYTHON_BIN — the resolved interpreter; re-resolve if it no longer fits.
# 2. VENV_DIR — see revalidate_existing_venv; a reused venv keeps its own
# interpreter, which re-resolving PYTHON_BIN does not change.
revalidate_python_constraint() {
if [[ "$ENABLE_PYFLINK" != "Yes" ]]; then
+ RECREATE_VENV_PATH=""
return 0
fi
@@ -1704,22 +1725,35 @@ revalidate_python_constraint() {
# PYTHON_BIN — is what pip runs under. After a version edit that lowers the
# Python ceiling, an existing venv built on a now-too-new interpreter would
# still be reused and fail at `pip install` time. Catch it at plan time:
-# offer to recreate it (deferred to setup_python_env via RECREATE_VENV, which
-# only ever deletes a directory carrying a pyvenv.cfg marker) or point at a
-# different path. Non-interactive callers get an actionable error instead of a
-# mid-install failure. No-op unless VENV_DIR is an existing venv.
+# offer to recreate it (deferred to setup_python_env via RECREATE_VENV_PATH,
+# which binds approval to this exact venv) or point at a different path.
+# Non-interactive callers get an actionable error instead of a mid-install
+# failure. No-op unless VENV_DIR is an existing venv.
revalidate_existing_venv() {
- RECREATE_VENV=0
- [[ "$ENABLE_PYFLINK" == "Yes" && -n "${VENV_DIR:-}" ]] || return 0
+ if [[ "$ENABLE_PYFLINK" != "Yes" || -z "${VENV_DIR:-}" ]]; then
+ RECREATE_VENV_PATH=""
+ return 0
+ fi
local kind
kind="$(validate_venv_dir "$VENV_DIR")" || true
- [[ "$kind" == "venv" ]] || return 0
+ if [[ "$kind" != "venv" ]]; then
+ RECREATE_VENV_PATH=""
+ return 0
+ fi
if validate_python_bin "$VENV_DIR/bin/python"; then
+ RECREATE_VENV_PATH=""
return 0
fi
+ # A previous edit may re-run plan validation. Preserve an approval only
+ # while it still refers to this same incompatible venv.
+ if [[ "${RECREATE_VENV_PATH:-}" == "$VENV_DIR" ]]; then
+ return 0
+ fi
+ RECREATE_VENV_PATH=""
+
local venv_pv
venv_pv="$("$VENV_DIR/bin/python" -c 'import sys; print(f"{sys.version_info.major}.{sys.version_info.minor}")' 2>/dev/null || printf 'unknown')"
ui_warn "Existing venv ${VENV_DIR} uses Python ${venv_pv}, incompatible with Flink Agents ${FLINK_AGENTS_VERSION} + Flink ${FLINK_VERSION} (needs >=3.10 and <3.$(python_minor_ceiling))"
@@ -1729,7 +1763,7 @@ revalidate_existing_venv() {
fi
if choose_install_method_interactive "Recreate ${VENV_DIR} with a compatible interpreter?"; then
- RECREATE_VENV=1
+ RECREATE_VENV_PATH="$VENV_DIR"
else
prompt_and_validate_venv_dir "$VENV_DIR"
# The newly chosen path may itself be an incompatible venv.
@@ -1991,4 +2025,3 @@ if [[ "${FLINK_AGENTS_INSTALL_SH_NO_RUN:-0}" != "1" ]]; then
configure_verbose
main
fi
-
diff --git a/tools/test/helpers/load.bash b/tools/test/helpers/load.bash
index e2cae738..f46f9891 100644
--- a/tools/test/helpers/load.bash
+++ b/tools/test/helpers/load.bash
@@ -28,7 +28,7 @@ reset_install_sh_state() {
ENABLE_PYFLINK="Ask"
PYTHON_BIN=""
PYFLINK_ACTUALLY_ENABLED=0
- RECREATE_VENV=0
+ RECREATE_VENV_PATH=""
FLINK_VERSION="2.2.0"
FLINK_AGENTS_VERSION="0.3.0"
FLINK_SCALA_VERSION="2.12"
diff --git a/tools/test/unit/edit_plan_quote.bats b/tools/test/unit/edit_plan_quote.bats
index 426b1d59..ac6ae60b 100644
--- a/tools/test/unit/edit_plan_quote.bats
+++ b/tools/test/unit/edit_plan_quote.bats
@@ -63,6 +63,7 @@ setup() {
VENV_DIR="/tmp/venv with space"
PYTHON_BIN="/usr/bin/python3"
FLINK_AGENTS_VERSION="0.2.0"
+ RECREATE_VENV_PATH="/tmp/venv with space"
local f="$BATS_TEST_TMPDIR/state"
edit_plan_dump_state "$f"
@@ -71,6 +72,7 @@ setup() {
INSTALL_FLINK=""; FLINK_VERSION=""; INSTALL_DIR=""; FLINK_HOME=""
FLINK_MAJOR_MINOR=""; ENABLE_PYFLINK=""; PYFLINK_ACTUALLY_ENABLED=0
VENV_DIR=""; PYTHON_BIN=""; FLINK_AGENTS_VERSION=""
+ RECREATE_VENV_PATH=""
# shellcheck disable=SC1090
source "$f"
@@ -85,4 +87,5 @@ setup() {
[ "$VENV_DIR" = "/tmp/venv with space" ]
[ "$PYTHON_BIN" = "/usr/bin/python3" ]
[ "$FLINK_AGENTS_VERSION" = "0.2.0" ]
+ [ "$RECREATE_VENV_PATH" = "/tmp/venv with space" ]
}
diff --git a/tools/test/unit/reconcile_plan_after_edit.bats b/tools/test/unit/reconcile_plan_after_edit.bats
new file mode 100644
index 00000000..afa4f864
--- /dev/null
+++ b/tools/test/unit/reconcile_plan_after_edit.bats
@@ -0,0 +1,55 @@
+#!/usr/bin/env bats
+
+setup() {
+ load '../helpers/load'
+ load '../helpers/shim'
+ load_install_sh
+ reset_install_sh_state
+ shim_setup
+ VENV_DIR="$BATS_TEST_TMPDIR/no-venv"
+}
+
+fake_python() {
+ local name="$1"
+ local ver="$2"
+ shim_bin_script "$name" "
+case \"\$*\" in
+ *'sys.version_info.major'*) echo '$ver' ;;
+esac
+"
+}
+
+@test "reconcile_plan_after_edit: re-detects FLINK_HOME and revalidates Python" {
+ local flink_home="$BATS_TEST_TMPDIR/flink-2.0.2"
+ mkdir -p "$flink_home/lib"
+ : > "$flink_home/lib/flink-dist-2.0.2.jar"
+ fake_python fake_py312 "3.12"
+ fake_python python3 "3.11"
+
+ INSTALL_FLINK="No"
+ FLINK_HOME="$flink_home"
+ FLINK_VERSION="2.2.1"
+ FLINK_MAJOR_MINOR="2.2"
+ ENABLE_PYFLINK="Yes"
+ FLINK_AGENTS_VERSION="0.3.0"
+ PYTHON_BIN="fake_py312"
+
+ reconcile_plan_after_edit flink_home
+
+ [ "$FLINK_VERSION" = "2.0.2" ]
+ [ "$FLINK_MAJOR_MINOR" = "2.0" ]
+ [ "$PYTHON_BIN" = "python3" ]
+}
+
+@test "reconcile_plan_after_edit: derives FLINK_HOME for managed installs" {
+ INSTALL_FLINK="Yes"
+ INSTALL_DIR="$BATS_TEST_TMPDIR/flink installs"
+ FLINK_VERSION="2.1.3"
+ FLINK_HOME="/stale/home"
+ FLINK_MAJOR_MINOR="2.2"
+
+ reconcile_plan_after_edit install_dir
+
+ [ "$FLINK_HOME" = "$INSTALL_DIR/flink-2.1.3" ]
+ [ "$FLINK_MAJOR_MINOR" = "2.1" ]
+}
diff --git a/tools/test/unit/revalidate_python_constraint.bats b/tools/test/unit/revalidate_python_constraint.bats
index 354df87a..96bbfa0a 100644
--- a/tools/test/unit/revalidate_python_constraint.bats
+++ b/tools/test/unit/revalidate_python_constraint.bats
@@ -158,11 +158,11 @@ EOF
ENABLE_PYFLINK="Yes"
FLINK_AGENTS_VERSION="0.2.1"
VENV_DIR="$BATS_TEST_TMPDIR/venv311"
- RECREATE_VENV=0
+ RECREATE_VENV_PATH=""
run revalidate_existing_venv
[ "$status" -eq 0 ]
- [ "$RECREATE_VENV" = "0" ]
+ [ "$RECREATE_VENV_PATH" = "" ]
}
@test "revalidate_existing_venv: no-op when VENV_DIR is not an existing venv" {
@@ -173,3 +173,72 @@ EOF
run revalidate_existing_venv
[ "$status" -eq 0 ]
}
+
+@test "revalidate_existing_venv: binds recreation approval to the current venv path" {
+ local old_venv="$BATS_TEST_TMPDIR/venv312"
+ fake_venv "$old_venv" "3.12"
+
+ ENABLE_PYFLINK="Yes"
+ FLINK_AGENTS_VERSION="0.2.1"
+ VENV_DIR="$old_venv"
+ is_promptable() { return 0; }
+ choose_install_method_interactive() { return 0; }
+
+ revalidate_existing_venv
+ [ "$RECREATE_VENV_PATH" = "$old_venv" ]
+}
+
+@test "revalidate_existing_venv: clears recreation approval after the venv path changes" {
+ local old_venv="$BATS_TEST_TMPDIR/venv312"
+ local new_venv="$BATS_TEST_TMPDIR/venv311"
+ fake_venv "$old_venv" "3.12"
+ fake_venv "$new_venv" "3.11"
+
+ ENABLE_PYFLINK="Yes"
+ FLINK_AGENTS_VERSION="0.2.1"
+ VENV_DIR="$new_venv"
+ RECREATE_VENV_PATH="$old_venv"
+
+ revalidate_existing_venv
+ [ "$RECREATE_VENV_PATH" = "" ]
+}
+
+@test "setup_python_env: does not apply recreation approval to a different venv" {
+ local approved_venv="$BATS_TEST_TMPDIR/approved-venv"
+ local selected_venv="$BATS_TEST_TMPDIR/selected-venv"
+ fake_venv "$approved_venv" "3.12"
+ fake_venv "$selected_venv" "3.11"
+ touch "$selected_venv/must-survive"
+
+ ENABLE_PYFLINK="Yes"
+ FLINK_AGENTS_VERSION="0.3.0"
+ FLINK_VERSION="2.0.2"
+ FLINK_MAJOR_MINOR="2.0"
+ VENV_DIR="$selected_venv"
+ RECREATE_VENV_PATH="$approved_venv"
+ create_venv() { fake_venv "$VENV_DIR" "3.11"; }
+ pip_install_quiet() { return 0; }
+
+ run setup_python_env
+ [ "$status" -eq 0 ]
+ [ -f "$selected_venv/must-survive" ]
+}
+
+@test "setup_python_env: recreates the exact venv path that was approved" {
+ local selected_venv="$BATS_TEST_TMPDIR/selected-venv"
+ fake_venv "$selected_venv" "3.12"
+ touch "$selected_venv/old-marker"
+
+ ENABLE_PYFLINK="Yes"
+ FLINK_AGENTS_VERSION="0.2.1"
+ FLINK_VERSION="2.0.2"
+ FLINK_MAJOR_MINOR="2.0"
+ VENV_DIR="$selected_venv"
+ RECREATE_VENV_PATH="$selected_venv"
+ create_venv() { fake_venv "$VENV_DIR" "3.11"; }
+ pip_install_quiet() { return 0; }
+
+ run setup_python_env
+ [ "$status" -eq 0 ]
+ [ ! -f "$selected_venv/old-marker" ]
+}
(flink-agents) 01/05: [dist] Add Flink 2.3 distribution and make it the default (#856)
Posted by we...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch release-0.3
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
commit c649ebe7734f718a85955fbafcdda3b65ac3497e
Author: Eugene <me...@gmail.com>
AuthorDate: Mon Jul 6 14:24:46 2026 +0800
[dist] Add Flink 2.3 distribution and make it the default (#856)
Add Flink 2.3 support and bump the default Flink version to 2.3.0. Central
2.3.0 jars are Java 11 bytecode, so the default still builds on JDK 11 — no
Java 17 requirement. dist modules follow a symmetric rule: the default
version (2.3) is bare and inherits the root flink.version; every other
version pins its own version + provided deps.
- dist: new flink-2.3 module; flink-2.2 now pins 2.2.1 + deps; root default 2.2.1 -> 2.3.0
- e2e: flink-2.3 default + flink-2.2 profile; e2e.sh default -> 2.3
- ci: add {flink 2.3, java 17} to it-java and it-python matrices
- python: bundle 2.3 thin jar into wheel manifest + build-backend test
- misc: ut.sh help, update_flink_version, faq/long_term_memory version notes
Verified: full build on JDK 11, all 5 dist versions (1.20/2.0/2.1/2.2/2.3) pass.
(cherry picked from commit 54bda560f4ff72c418b1c65c7ea15662ff9d6c30)
---
.github/workflows/ci.yml | 6 +++
dist/flink-2.2/pom.xml | 44 ++++++++++++++++++++++
dist/{flink-2.2 => flink-2.3}/pom.xml | 4 +-
dist/pom.xml | 2 +
.../docs/development/memory/long_term_memory.md | 2 +-
docs/content/docs/faq/faq.md | 2 +-
.../pom.xml | 14 ++++++-
pom.xml | 2 +-
python/_build_backend/tests/test_backend.py | 7 ++++
python/jar_manifest.json | 6 +++
tools/e2e.sh | 2 +-
tools/ut.sh | 4 +-
12 files changed, 85 insertions(+), 10 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 198c1452..6d27d2ec 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -171,6 +171,9 @@ jobs:
- os: 'ubuntu-latest'
python-version: '3.12'
flink-version: "2.2"
+ - os: 'ubuntu-latest'
+ python-version: '3.12'
+ flink-version: "2.3"
steps:
- uses: actions/checkout@v4
- name: Install java
@@ -216,6 +219,9 @@ jobs:
- os: 'ubuntu-latest'
java-version: "21"
flink-version: "2.2"
+ - os: 'ubuntu-latest'
+ java-version: "17"
+ flink-version: "2.3"
steps:
- uses: actions/checkout@v4
- name: Install java
diff --git a/dist/flink-2.2/pom.xml b/dist/flink-2.2/pom.xml
index 4cdb742f..151d2eaf 100644
--- a/dist/flink-2.2/pom.xml
+++ b/dist/flink-2.2/pom.xml
@@ -28,6 +28,50 @@ under the License.
<artifactId>flink-agents-dist-flink-2.2</artifactId>
<name>Flink Agents : Dist : Flink 2.2</name>
+ <properties>
+ <flink.version>${flink.2.2.version}</flink.version>
+ </properties>
+
+ <dependencies>
+ <!-- Flink dependencies -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-clients</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-api-java</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-api-java-bridge</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-python</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+
<build>
<plugins>
<plugin>
diff --git a/dist/flink-2.2/pom.xml b/dist/flink-2.3/pom.xml
similarity index 97%
copy from dist/flink-2.2/pom.xml
copy to dist/flink-2.3/pom.xml
index 4cdb742f..8deaaf88 100644
--- a/dist/flink-2.2/pom.xml
+++ b/dist/flink-2.3/pom.xml
@@ -25,8 +25,8 @@ under the License.
<version>0.3-SNAPSHOT</version>
</parent>
- <artifactId>flink-agents-dist-flink-2.2</artifactId>
- <name>Flink Agents : Dist : Flink 2.2</name>
+ <artifactId>flink-agents-dist-flink-2.3</artifactId>
+ <name>Flink Agents : Dist : Flink 2.3</name>
<build>
<plugins>
diff --git a/dist/pom.xml b/dist/pom.xml
index 0a72542a..1227898d 100644
--- a/dist/pom.xml
+++ b/dist/pom.xml
@@ -34,6 +34,7 @@ under the License.
<flink.2.0.version>2.0.2</flink.2.0.version>
<flink.2.1.version>2.1.3</flink.2.1.version>
<flink.2.2.version>2.2.1</flink.2.2.version>
+ <flink.2.3.version>2.3.0</flink.2.3.version>
</properties>
<modules>
@@ -42,6 +43,7 @@ under the License.
<module>flink-2.0</module>
<module>flink-2.1</module>
<module>flink-2.2</module>
+ <module>flink-2.3</module>
</modules>
<dependencies>
diff --git a/docs/content/docs/development/memory/long_term_memory.md b/docs/content/docs/development/memory/long_term_memory.md
index eeb4c79d..6de50651 100644
--- a/docs/content/docs/development/memory/long_term_memory.md
+++ b/docs/content/docs/development/memory/long_term_memory.md
@@ -35,7 +35,7 @@ Declare the following resources in your agent plan:
- An [EmbeddingModel]({{< ref "docs/development/embedding_models" >}}) for vector generation
- A [VectorStore]({{< ref "docs/development/vector_stores" >}}) for persistent storage
-> **Java prerequisite:** Mem0 invokes the chat and embedding models on its own thread executor, which relies on the async-friendly pemja fix. When Mem0 Long-Term Memory is configured together with Java actions, the runtime requires a Flink version of `1.20.5`, `2.0.2`, `2.1.3`, `2.2.1` or higher; otherwise it throws at startup. Either upgrade Flink or use the Python API.
+> **Java prerequisite:** Mem0 invokes the chat and embedding models on its own thread executor, which relies on the async-friendly pemja fix. When Mem0 Long-Term Memory is configured together with Java actions, the runtime requires a Flink version of `1.20.5`, `2.0.2`, `2.1.3`, `2.2.1`, `2.3.0` or higher; otherwise it throws at startup. Either upgrade Flink or use the Python API.
## Configuration
diff --git a/docs/content/docs/faq/faq.md b/docs/content/docs/faq/faq.md
index 76ed3e03..bc0cb279 100644
--- a/docs/content/docs/faq/faq.md
+++ b/docs/content/docs/faq/faq.md
@@ -87,7 +87,7 @@ This is important because:
- **For Java users on JDK 21+**: Async execution is available.
- **For Java users on JDK < 21**: Async execution is not available and falls back to synchronous execution.
-> **Cross-language async note**: Async execution for cross-language resources requires the pemja 0.5.7 fix, available in Flink 1.20.5+ / 2.0.2+ / 2.1.3+ / 2.2.1+. Current builds target Flink 2.2.0, so cross-language calls still run synchronously for now; this is resolved automatically once running on a Flink version that includes the fix.
+> **Cross-language async note**: Async execution for cross-language resources requires the pemja 0.5.7 fix, available in Flink 1.20.5+ / 2.0.2+ / 2.1.3+ / 2.2.1+ / 2.3+. Current builds target Flink 2.3.0, which includes the fix, so cross-language async is enabled by default; on older Flink versions it falls back to synchronous execution automatically.
### Native Integration Support Matrix
diff --git a/e2e-test/flink-agents-end-to-end-tests-integration/pom.xml b/e2e-test/flink-agents-end-to-end-tests-integration/pom.xml
index bfbdf1e6..479a47ba 100644
--- a/e2e-test/flink-agents-end-to-end-tests-integration/pom.xml
+++ b/e2e-test/flink-agents-end-to-end-tests-integration/pom.xml
@@ -33,9 +33,10 @@ under the License.
<flink.2.0.version>2.0.2</flink.2.0.version>
<flink.2.1.version>2.1.3</flink.2.1.version>
<flink.2.2.version>2.2.1</flink.2.2.version>
+ <flink.2.3.version>2.3.0</flink.2.3.version>
- <flink.version>${flink.2.2.version}</flink.version>
- <flink.agents.dist.artifactId>flink-agents-dist-flink-2.2</flink.agents.dist.artifactId>
+ <flink.version>${flink.2.3.version}</flink.version>
+ <flink.agents.dist.artifactId>flink-agents-dist-flink-2.3</flink.agents.dist.artifactId>
</properties>
<dependencies>
@@ -199,6 +200,15 @@ under the License.
<flink.agents.dist.artifactId>flink-agents-dist-flink-2.1</flink.agents.dist.artifactId>
</properties>
</profile>
+
+ <!-- Flink 2.2 Profile -->
+ <profile>
+ <id>flink-2.2</id>
+ <properties>
+ <flink.version>${flink.2.2.version}</flink.version>
+ <flink.agents.dist.artifactId>flink-agents-dist-flink-2.2</flink.agents.dist.artifactId>
+ </properties>
+ </profile>
</profiles>
</project>
diff --git a/pom.xml b/pom.xml
index 4728fa58..108841ac 100644
--- a/pom.xml
+++ b/pom.xml
@@ -41,7 +41,7 @@ under the License.
<maven.compiler.target>${target.java.version}</maven.compiler.target>
<spotless.version>2.27.1</spotless.version>
<spotless.skip>false</spotless.skip>
- <flink.version>2.2.1</flink.version>
+ <flink.version>2.3.0</flink.version>
<kafka.version>4.0.0</kafka.version>
<fluss.version>0.9.0-incubating</fluss.version>
<junit5.version>5.10.1</junit5.version>
diff --git a/python/_build_backend/tests/test_backend.py b/python/_build_backend/tests/test_backend.py
index a8fc9b6d..9bec750f 100644
--- a/python/_build_backend/tests/test_backend.py
+++ b/python/_build_backend/tests/test_backend.py
@@ -180,6 +180,7 @@ def maven_server(tmp_path_factory) -> tuple[str, Path]:
("flink-agents-dist-flink-2.0", "thin"),
("flink-agents-dist-flink-2.1", "thin"),
("flink-agents-dist-flink-2.2", "thin"),
+ ("flink-agents-dist-flink-2.3", "thin"),
]
for artifact_id, classifier in artifacts:
@@ -243,6 +244,12 @@ def test_download_jars_from_local_server(tmp_path, monkeypatch, maven_server) ->
"dest": "flink_agents/lib/flink-2.2/",
"sha256": expected_sha,
},
+ {
+ "artifact_id": "flink-agents-dist-flink-2.3",
+ "classifier": "thin",
+ "dest": "flink_agents/lib/flink-2.3/",
+ "sha256": expected_sha,
+ },
],
}
_write_manifest(tmp_path / "jar_manifest.json", manifest)
diff --git a/python/jar_manifest.json b/python/jar_manifest.json
index 5077c4d1..4ee74ddc 100644
--- a/python/jar_manifest.json
+++ b/python/jar_manifest.json
@@ -31,6 +31,12 @@
"classifier": "thin",
"dest": "flink_agents/lib/flink-2.2/",
"sha256": "PLACEHOLDER"
+ },
+ {
+ "artifact_id": "flink-agents-dist-flink-2.3",
+ "classifier": "thin",
+ "dest": "flink_agents/lib/flink-2.3/",
+ "sha256": "PLACEHOLDER"
}
]
}
diff --git a/tools/e2e.sh b/tools/e2e.sh
index bc04762e..5a5c7713 100755
--- a/tools/e2e.sh
+++ b/tools/e2e.sh
@@ -15,7 +15,7 @@
# limitations under the License.
#
-DEFAULT_FLINK_VERSION="2.2"
+DEFAULT_FLINK_VERSION="2.3"
function run_test {
local description="$1"
diff --git a/tools/ut.sh b/tools/ut.sh
index a40303e4..76876c79 100755
--- a/tools/ut.sh
+++ b/tools/ut.sh
@@ -43,8 +43,8 @@ Options:
-e, --e2e Run e2e tests
-b, --both Run both Java and Python tests (default)
-f, --flink Specify Flink version to test (can be used multiple times)
- Supported versions: 2.2, 1.20
- Examples: -f 2.2, -f 1.20, -f 2.2 -f 1.20
+ Supported versions: 2.3, 2.2, 2.1, 2.0, 1.20
+ Examples: -f 2.3, -f 1.20, -f 2.3 -f 1.20
Default: run all versions if not specified
-v, --verbose Show verbose output
-h, --help Display this help message