purushah commented on code in PR #1094:
URL: https://github.com/apache/flink-agents/pull/1094#discussion_r3969517838
##########
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:
Great catch — you're right, and my compatibility note overclaimed.
PythonFunction hashes stable strings (module, qualName), so Python-action keys
were already stable across restarts and this change does invalidate their
existing state. Will correct the note: Java action state was unrecoverable
across restarts (the bug — no regression possible); Python action state was
recoverable, and the first recovery after upgrading re-executes those actions
once and re-persists under the new format. I'll spell that out as the upgrade
guidance. If you'd prefer a legacy-key fallback read for Python actions during
a deprecation window I'm happy to add one, though given the project's pre-1.0
status a release note may be the better trade.
##########
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:
Agreed — a fresh topic only avoids stale entries; restore from an older
checkpoint re-executes either way since the saved results are unreadable under
the new format. Will fold that into the note.
##########
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:
Right — one JVM shares the same Class objects, so that test can't
discriminate. Will add a cross-JVM test that launches two separate java
processes, prints the generated key for the same action in each, and asserts
they match — it fails on the pre-fix code and keeps the in-JVM tests as fast
pins.
--
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]
