purushah commented on code in PR #1094:
URL: https://github.com/apache/flink-agents/pull/1094#discussion_r3964462917
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtil.java:
##########
@@ -222,8 +222,11 @@ private static String generateUUIDForEvent(Event event)
throws IOException {
}
private static String generateUUIDForAction(Action action) throws
IOException {
+ // Action.hashCode() folds in JavaFunction's Class[] parameterTypes,
and Class.hashCode()
+ // is the per-JVM identity hash — so the hash-derived UUID changes on
every process
+ // restart and recovery lookups can never hit. Derive from the
plan-unique action name,
+ // which is stable across restarts.
return String.valueOf(
- UUID.nameUUIDFromBytes(
-
String.valueOf(action.hashCode()).getBytes(StandardCharsets.UTF_8)));
+
UUID.nameUUIDFromBytes(action.getName().getBytes(StandardCharsets.UTF_8)));
Review Comment:
Good catch, thanks. `PythonFunction` hashes `module` and `qualName`, so
Python action keys were already stable and this change does alter them. I'll
fix the compatibility note: a Python action in flight at the restoring
checkpoint re-executes once after upgrading. To avoid that, drain in-flight
work before the stop-with-savepoint, or start the upgraded job with a fresh
action-state topic/table. No fallback to the old key, per the beta
breaking-change policy. I'll add a note to `deployment.md`.
##########
runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtilTest.java:
##########
@@ -116,6 +118,43 @@ public void
testGenerateKeyRejectsNonPositiveMaxParallelism() throws Exception {
() -> ActionStateUtil.generateKey(key, 1, action, inputEvent,
-1));
}
+ /**
+ * The action-UUID key segment must be derived from the plan-unique action
NAME, never from
+ * {@code Action.hashCode()}: the hash folds in {@code Class.hashCode()}
(a per-JVM identity
+ * hash), so a hash-derived segment silently changes across process
restarts and recovery
+ * lookups can never hit. This pins the derivation so any future change to
the key format is a
+ * conscious, reviewed break of cross-restart state compatibility.
+ */
+ @Test
+ public void testActionUUIDSegmentDerivesFromActionName() throws Exception {
+ Action action = new NoOpAction("test-action");
+ String generatedKey =
+ ActionStateUtil.generateKey(
+ "test-key", 1, action, new InputEvent("test-input"),
MAX_PARALLELISM);
+
+ String actionUUIDSegment =
ActionStateUtil.parseKey(generatedKey).get(3);
+ assertEquals(
+
UUID.nameUUIDFromBytes("test-action".getBytes(StandardCharsets.UTF_8)).toString(),
+ 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 =
+ ActionStateUtil.generateKey(
+ "test-key", 7, new NoOpAction("stable-name"), event,
MAX_PARALLELISM);
+ String second =
+ ActionStateUtil.generateKey(
+ "test-key", 7, new NoOpAction("stable-name"), event,
MAX_PARALLELISM);
+ assertEquals(first, second);
Review Comment:
Agreed, that test passes before the fix too. I'll replace it with a test
that loads an `Event` subclass through two isolating class loaders, so the
parameter `Class` objects get different identity hashes, which is the same
effect a JVM restart has on `JavaFunction.hashCode()`. Verified it fails on the
pre-fix code and passes with the fix.
--
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]
