weiqingy commented on code in PR #885:
URL: https://github.com/apache/flink-agents/pull/885#discussion_r3840641935
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java:
##########
@@ -175,10 +181,11 @@ public ActionState get(Object key, long seqNum, Action
action, Event event) thro
// the requested seqNum
return stateSeqNum > seqNum;
Review Comment:
The Fluss `get()` test you just added looks like it would fail if it were
pointed at Kafka. Copying `FlussActionStateStoreTest.java:145-152` into
`KafkaActionStateStoreTest`: `get("a_1", 0L, ...)` misses, this `removeIf`
parses the cached `a_1_<eventUuid>_<actionUuid>` (Flink key `a`, seq 1) into 4
parts, `1 > 0` fires, and the `containsKey` assertion fails. Kafka has the
`pruneState` twin at `KafkaActionStateStoreTest.java:261-274`, but not this one.
The reason looks like this scan isn't tied to a key at all. There's no
prefix filter and no `parts.get(0)` check, just `stateSeqNum > seqNum` applied
to every entry in the map. One map holds every key the subtask owns (`:82`,
store created per operator at `DurableExecutionManager.java:118-125`), and
sequence numbers count per key (`DurableExecutionManager.java:233`), so one
key's lookup can drop another key's state.
The cache isn't the durable record, so how much that costs depends on when
the evicted key is read next. The case I can't rule out is just after a
restore, when the replay path reads the rebuilt cache:
`processActionTaskForKey` skips execution only when `actionState != null &&
actionState.isCompleted()` (`ActionExecutionOperator.java:435`), so an entry
that got dropped runs again. That needs action tasks for different keys to
interleave during replay, which I haven't reproduced, so I'd call it a hazard
worth checking rather than a break.
To be clear, this predicate is base code and not something the PR
introduced. The only change inside the block was the catch clause and its log
line, and the other three scan sites do carry the guard at head (`:294-301`
here, and `FlussActionStateStore.java:249-256` for both of its callers). Would
the same `parts.get(0)` check you just added on Fluss fit here as well, or
would you rather it went out separately?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java:
##########
@@ -250,10 +251,11 @@ private void removeStateEntries(String keyPrefix,
LongPredicate seqNumFilter) {
}
try {
List<String> parts =
ActionStateUtil.parseKey(entry.getKey());
- if (parts.size() >= 2) {
- long stateSeqNum =
Long.parseLong(parts.get(1));
- return seqNumFilter.test(stateSeqNum);
+ if (!parts.get(0).equals(key)) {
+ return false;
}
+ long stateSeqNum =
Long.parseLong(parts.get(1));
+ return seqNumFilter.test(stateSeqNum);
} catch (Exception e) {
LOG.warn("Failed to parse state key: {}",
entry.getKey(), e);
Review Comment:
nit: the trailing `e` with a single `{}` means this logs a full stack trace,
and it's the Fluss twin of the Kafka site you just demoted. A Flink key
containing `_` never parses, so `:262` keeps the entries and `get()` runs this
scan again on every new sequence number for that key, which repeats the trace
for as long as the job lives.
It's narrower than the Kafka one, since this scan is prefix-filtered so only
that key's own entries reach it, and the docs now describe the retention as
expected. Would the same DEBUG treatment fit here, and is the stack trace
earning its place on this path?
--
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]
