da-daken commented on code in PR #1024:
URL: https://github.com/apache/flink-agents/pull/1024#discussion_r3860124249
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtil.java:
##########
@@ -58,10 +83,110 @@ public static String generateKey(
public static List<String> parseKey(String key) {
Preconditions.checkNotNull(key, "key cannot be null.");
String[] parts = key.split(KEY_SEPARATOR);
- Preconditions.checkArgument(parts.length == 4, "Invalid key format.");
+ Preconditions.checkArgument(parts.length == KEY_SEGMENT_COUNT,
"Invalid key format.");
return List.of(parts);
}
+ /**
+ * Extracts the key-group from a composite state key. The key-group is the
first segment and was
+ * computed from the original typed key via {@link
KeyGroupRangeAssignment#assignToKeyGroup}.
+ * Rejects keys without the expected segment layout, including keys
written in the pre-key-group
+ * 4-segment format.
+ */
+ public static int parseKeyGroup(String key) {
+ Preconditions.checkNotNull(key, "key cannot be null.");
+ String[] parts = key.split(KEY_SEPARATOR);
+ Preconditions.checkArgument(parts.length == KEY_SEGMENT_COUNT,
"Invalid key format.");
+ return Integer.parseInt(parts[KEY_GROUP_SEGMENT]);
+ }
+
+ /**
+ * Returns {@code true} when {@code stateKey} has the expected segment
layout and its
+ * business-key segment equals {@code businessKey}. Comparison is
segment-exact; substring
+ * matching is deliberately avoided because a numeric business key can
collide with another
+ * record's sequence-number segment.
+ */
+ public static boolean matchesBusinessKey(String stateKey, Object
businessKey) {
+ String[] parts = stateKey.split(KEY_SEPARATOR);
+ return parts.length == KEY_SEGMENT_COUNT
+ && parts[BUSINESS_KEY_SEGMENT].equals(businessKey.toString());
+ }
+
+ /** Like {@link #matchesBusinessKey} with an additional exact
sequence-number segment match. */
+ public static boolean matchesBusinessKeyAndSeqNum(
+ String stateKey, Object businessKey, long seqNum) {
+ String[] parts = stateKey.split(KEY_SEPARATOR);
+ return parts.length == KEY_SEGMENT_COUNT
+ && parts[BUSINESS_KEY_SEGMENT].equals(businessKey.toString())
+ && parts[SEQ_NUM_SEGMENT].equals(String.valueOf(seqNum));
+ }
+
+ /**
+ * Like {@link #matchesBusinessKey} with an additional predicate over the
parsed sequence-number
+ * segment. Returns {@code false} for keys that cannot be attributed
(malformed layout or
+ * unparsable sequence number): never prune what cannot be attributed.
+ */
+ public static boolean matchesBusinessKeyWithSeqNum(
+ String stateKey, Object businessKey, LongPredicate seqNumFilter) {
+ String[] parts = stateKey.split(KEY_SEPARATOR);
+ if (parts.length != KEY_SEGMENT_COUNT
+ ||
!parts[BUSINESS_KEY_SEGMENT].equals(businessKey.toString())) {
+ return false;
+ }
+ try {
+ return seqNumFilter.test(Long.parseLong(parts[SEQ_NUM_SEGMENT]));
+ } catch (NumberFormatException e) {
+ LOG.warn("Failed to parse sequence number from state key: {}",
stateKey);
+ return false;
+ }
+ }
+
+ /**
+ * Returns {@code true} if the composite {@code stateKey}'s key-group is
accepted by the given
+ * ownership filter. A {@code null} filter retains every key (the default
for in-memory and test
+ * backends).
+ *
+ * <p>Keys without the expected 5-segment layout — including records
written in the
+ * pre-key-group 4-segment format — have UNKNOWN ownership: they cannot be
attributed to a
+ * key-group, so they are retained in every subtask rather than dropped.
This preserves durable
+ * state across a key-group upgrade at the cost of a bounded, one-time
memory amplification for
+ * the legacy recovery tail, which ages out once a new checkpoint marker
advances past those
+ * records. Lookups still find such records via {@link #legacyKeyOf}. A
5-segment key whose
+ * key-group segment fails to parse is likewise retained as a fail-safe.
+ */
+ public static boolean isKeyRetained(@Nullable IntPredicate
ownershipFilter, String stateKey) {
+ if (ownershipFilter == null) {
+ return true;
+ }
+ String[] parts = stateKey.split(KEY_SEPARATOR);
Review Comment:
Good catch! I have set the new stateKey to:
keyGroup_seqNum_eventUUID_actionUUID_businessKey. The first four parts are in
fixed formats and will not be affected by the underscore (_) delimiter.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java:
##########
@@ -105,12 +106,20 @@ public class FlussActionStateStore implements
ActionStateStore {
/** In-memory cache for O(1) state lookups; rebuilt from Fluss log on
recovery. */
private final Map<String, ActionState> actionStates;
+ // When set, only records whose key-group is accepted by this predicate
are kept in the
+ // in-memory cache during rebuildState; null means retain all keys
(default).
+ private IntPredicate ownershipFilter;
+
+ // The operator's maximum parallelism, used to compute key-groups
consistently with Flink.
+ private int maxParallelism;
Review Comment:
Thank you for pointing that out. The fact that FlussActionStateStoreIT
didn't run successfully caused me to miss this point. It has been fixed in the
latest commit.
--
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]
