rob-9 commented on code in PR #885:
URL: https://github.com/apache/flink-agents/pull/885#discussion_r3930832570
##########
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:
Addressed after merging `main`: Kafka cleanup now uses the shared
segment-exact matcher, so the reported `a` / `a_1` case no longer removes
another key's state. I also added coverage for the cache-miss cleanup path.
##########
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:
Addressed after merging `main`: underscore-containing keys now parse
correctly, and the shared matcher no longer emits the repeated full-stack-trace
warning described here.
--
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]
