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 b2aa864d [runtime] Cover the cross-loader action key failure and
document action renaming (#1094)
b2aa864d is described below
commit b2aa864dd26ecbf4dc6c844156ec4d7e2c7b7318
Author: puru <[email protected]>
AuthorDate: Mon Sep 14 23:32:18 2026 -0700
[runtime] Cover the cross-loader action key failure and document action
renaming (#1094)
Generated-by: Claude Code (Claude Fable 5.1)
Co-authored-by: purshotam shah <[email protected]>
Co-authored-by: Claude Fable 5.1 <[email protected]>
---
docs/content/docs/operations/deployment.md | 4 +-
.../ActionStateUtilCrossClassLoaderTest.java | 124 +++++++++++++++++++++
.../runtime/actionstate/ActionStateUtilTest.java | 15 ---
3 files changed, 127 insertions(+), 16 deletions(-)
diff --git a/docs/content/docs/operations/deployment.md
b/docs/content/docs/operations/deployment.md
index e5c59e86..8ff65b1b 100644
--- a/docs/content/docs/operations/deployment.md
+++ b/docs/content/docs/operations/deployment.md
@@ -103,10 +103,12 @@ The same persisted action state is also used by
fine-grained durable execution.
{{< /hint >}}
{{< hint warning >}}
-The action-state key format has changed and existing action-state records are
unsupported. When upgrading, use a fresh Kafka topic or Fluss table and start
without an older checkpoint or savepoint. Recovery validates the current
fields, but cannot reliably distinguish every older record from the current
format.
+The action-state key format changed in Flink Agents 0.4 and action-state
records written by earlier versions are unsupported. When upgrading, use a
fresh Kafka topic or Fluss table and start without an older checkpoint or
savepoint. Recovery validates the current fields, but cannot reliably
distinguish every older record from the current format.
Action-state keys contain a digest of the serialized key. Changing key types
or serializer configuration when recovering existing state is unsupported, even
if Flink accepts the change for keyed state: the same key can serialize to
different bytes, causing recovery to miss completed actions and repeat their
side effects. These changes are not detected by the action-state store.
Preserve the original key types and serializer configuration for recovery.
Custom key serializers must produce [...]
+Action-state keys identify an action by its name. Keeping an action's name
across a code change lets recovery reuse its completed results; renaming an
action makes its recorded results unreachable, so it is executed again on the
next recovery.
+
Dedicate each Kafka topic or Fluss table to one logical Flink Agents operator,
shared by that operator's subtasks. Action-state keys do not contain a job or
operator namespace, so sharing a backend between logical operators can allow
otherwise identical records to collide.
{{< /hint >}}
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtilCrossClassLoaderTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtilCrossClassLoaderTest.java
new file mode 100644
index 00000000..71afe60b
--- /dev/null
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtilCrossClassLoaderTest.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.runtime.actionstate;
+
+import org.apache.flink.agents.api.Event;
+import org.apache.flink.agents.api.InputEvent;
+import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.plan.JavaFunction;
+import org.apache.flink.agents.plan.actions.Action;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.List;
+
+import static
org.apache.flink.agents.runtime.actionstate.ActionStateTestUtils.generateKey;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/**
+ * Reproduces the cross-restart failure of the action-UUID key segment inside
one JVM.
+ *
+ * <p>{@link JavaFunction#hashCode()} folds in {@code
Arrays.hashCode(Class<?>[] parameterTypes)},
+ * and {@link Class#hashCode()} is an identity hash, so the same action hashes
differently in every
+ * process. A fresh JVM cannot be started from a unit test, but a fresh class
loader produces the
+ * same effect: it defines a distinct {@link Class} object for the same bytes,
with its own identity
+ * hash. The loader below defines only {@link ParamEvent} itself and delegates
everything else to
+ * the parent, so {@link Action}'s signature check against {@link Event} still
passes.
+ */
+public class ActionStateUtilCrossClassLoaderTest {
+
+ private static final int MAX_PARALLELISM = 128;
+
+ /** Subclass of {@link Event} that the isolating loader defines a second
time. */
+ public static class ParamEvent extends Event {
+ public ParamEvent() {
+ super("param");
+ }
+ }
+
+ @Test
+ public void
testKeyIsStableWhenParameterClassesComeFromDifferentClassLoaders()
+ throws Exception {
+ Class<?> paramFromLoaderA = new
IsolatingLoader().loadClass(ParamEvent.class.getName());
+ Class<?> paramFromLoaderB = new
IsolatingLoader().loadClass(ParamEvent.class.getName());
+ assertNotSame(paramFromLoaderA, paramFromLoaderB);
+
+ Action first = actionWithParameterType(paramFromLoaderA);
+ Action second = actionWithParameterType(paramFromLoaderB);
+ // Identity hashes almost always differ; this is the pre-fix failure
mode. On the rare
+ // collision there is nothing to test, so skip rather than fail.
+ assumeTrue(
+ first.hashCode() != second.hashCode(),
+ "identity hashes of the two Class objects collided");
+
+ InputEvent event = new InputEvent("test-input");
+ assertEquals(
+ generateKey("test-key", 7, first, event, MAX_PARALLELISM),
+ generateKey("test-key", 7, second, event, MAX_PARALLELISM));
+ }
+
+ private static Action actionWithParameterType(Class<?> eventParameterType)
throws Exception {
+ return new Action(
+ "stable-name",
+ new JavaFunction(
+ NoOpAction.class.getName(),
+ "doNothing",
+ new Class<?>[] {eventParameterType,
RunnerContext.class}),
+ List.of(InputEvent.EVENT_TYPE));
+ }
+
+ /** Defines {@link ParamEvent} itself and delegates every other class to
the parent loader. */
+ private static final class IsolatingLoader extends ClassLoader {
+ IsolatingLoader() {
+ super(ActionStateUtilCrossClassLoaderTest.class.getClassLoader());
+ }
+
+ @Override
+ protected Class<?> loadClass(String name, boolean resolve) throws
ClassNotFoundException {
+ if (!name.equals(ParamEvent.class.getName())) {
+ return super.loadClass(name, resolve);
+ }
+ synchronized (getClassLoadingLock(name)) {
+ Class<?> loaded = findLoadedClass(name);
+ if (loaded == null) {
+ byte[] bytes = readClassBytes(name);
+ loaded = defineClass(name, bytes, 0, bytes.length);
+ }
+ if (resolve) {
+ resolveClass(loaded);
+ }
+ return loaded;
+ }
+ }
+
+ private byte[] readClassBytes(String name) throws
ClassNotFoundException {
+ String resource = name.replace('.', '/') + ".class";
+ try (InputStream in = getParent().getResourceAsStream(resource)) {
+ if (in == null) {
+ throw new ClassNotFoundException(name);
+ }
+ return in.readAllBytes();
+ } catch (IOException e) {
+ throw new ClassNotFoundException(name, e);
+ }
+ }
+ }
+}
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtilTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtilTest.java
index deff9f67..2430784d 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtilTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtilTest.java
@@ -182,21 +182,6 @@ public class ActionStateUtilTest {
actionUUIDSegment);
}
- /**
- * Two separately constructed Action instances with the same name — which
is what "the same
- * action, after a JVM restart" looks like — must produce identical state
keys, or recovery can
- * never replay.
- */
- @Test
- public void testSameActionNameYieldsSameKeyAcrossInstances() throws
Exception {
- InputEvent event = new InputEvent("test-input");
- String first =
- generateKey("test-key", 7, new NoOpAction("stable-name"),
event, MAX_PARALLELISM);
- String second =
- generateKey("test-key", 7, new NoOpAction("stable-name"),
event, MAX_PARALLELISM);
- assertEquals(first, second);
- }
-
@Test
public void testParseKeyValidKey() throws Exception {
// Create test data and generate a key
