GreatEugenius commented on code in PR #1024:
URL: https://github.com/apache/flink-agents/pull/1024#discussion_r3817886042
##########
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:
The ownership check hashes the String parsed from the durable state key, but
Flink assigned keyed-state ownership using the original typed key.
KeyGroupRangeAssignment hashes different object types differently; for example,
with max parallelism 128, Long(1) maps to key-group 86 while String("1") maps
to 54. The true owner can therefore discard its recovered ActionState while
another subtask retains it, allowing replay to execute an already completed
action again. Please persist the key-group computed from the original typed key
in the WAL record and compare that value with currentSubtaskKeyGroupRange,
instead of reconstructing ownership from the string form.
--
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]
