GreatEugenius commented on code in PR #1024:
URL: https://github.com/apache/flink-agents/pull/1024#discussion_r3827537938
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtil.java:
##########
@@ -62,6 +68,25 @@ public static List<String> parseKey(String key) {
return List.of(parts);
}
+ /**
+ * Returns {@code true} if the composite {@code stateKey}'s business key
should be retained in a
+ * subtask's in-memory cache under the given ownership filter. A {@code
null} filter retains
+ * every key (the default for in-memory and test backends). If the key
cannot be parsed, it is
+ * retained as a fail-safe: prefer keeping a valid key over dropping it on
a parse error.
+ */
+ public static boolean isKeyRetained(
+ @Nullable Predicate<String> ownershipFilter, String stateKey) {
+ if (ownershipFilter == null) {
+ return true;
+ }
+ try {
+ return ownershipFilter.test(parseKey(stateKey).get(0));
Review Comment:
Thanks for the detailed analysis. Persisting the key-group at write time
makes sense.
I am concerned about dropping all legacy 4-segment records during upgrade. A
missing key-group does not have to mean either dropping the record or falling
back to the incorrect string hash. We could treat it as UNKNOWN ownership:
Records with a key-group: filter normally using the current subtask’s
KeyGroupRange.
Legacy records without a key-group: temporarily retain them in every subtask.
This preserves the old memory amplification only for the legacy recovery
tail during the first upgraded attempt, while avoiding the loss of durable
state. Once a new checkpoint marker advances past those records, they will no
longer be included in the next recovery tail.
Dropping legacy records may re-execute every affected action or durable call
in the recovery tail, potentially repeating external side effects. For durable
execution, I think the bounded one-time memory overhead is preferable to
weakening recovery correctness.
--
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]
