wenjin272 commented on code in PR #1024:
URL: https://github.com/apache/flink-agents/pull/1024#discussion_r3842891090
##########
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:
`maxParallelism` remains `0` when the public `FlussActionStateStore(config)`
constructor is used, and the Kafka constructor has the same issue. Any direct
`put()` or `get()` then fails in `generateKey()` because it requires a positive
value. I reproduced this by running `FlussActionStateStoreIT` explicitly: 10 of
its 11 tests fail with `maxParallelism must be positive but was 0`. Since API
compatibility is not required during beta, could we make `maxParallelism` a
required constructor argument and final, rather than relying on a later mutable
setter?
##########
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:
Could we avoid using the number of underscore-separated segments to parse
the state key? `businessKey` is inserted without escaping, so a key such as
`tenant_user` produces more than five segments and is treated as UNKNOWN,
causing every subtask to retain it and leaving the orphan-state leak unfixed.
Since the project is still beta and previous ActionState data does not need to
be preserved, I suggest defining a single new, unambiguous key format and
removing the legacy compatibility logic entirely. For example, the business key
could be length-prefixed, or the fixed fields could be parsed from both ends.
##########
runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreIT.java:
##########
@@ -206,6 +206,44 @@ void testRebuildStateWithRecoveryMarkers() throws
Exception {
}
}
+ /**
+ * Reproduces the orphan-state leak fix: after recovery, a subtask must
keep only the keys it
+ * owns and drop keys owned by other subtasks. Here "A" is owned and "B"
is foreign, so the
+ * rebuilt cache must contain "A" but not "B".
+ */
+ @Test
+ @SuppressWarnings("unchecked")
+ void testRebuildStateFiltersForeignKeys() throws Exception {
Review Comment:
This regression test is not executed by the current CI. The class name ends
with `IT`, while the repository runs ordinary `mvn test` with Surefire default
discovery patterns, which do not include `*IT`. Could we either rename this
class to something ending in `Test`, such as
`FlussActionStateStoreIntegrationTest`, or explicitly wire `*IT` tests into
Surefire/Failsafe and CI? Once enabled, the `maxParallelism=0` failure above
will also need to be fixed.
--
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]
