da-daken commented on issue #1302:
URL: https://github.com/apache/flink-agents/issues/1302#issuecomment-6340817529
hi @wenjin272 , Thank you for your detailed description, The issue where
event interleaving then causes C to execute should not exist. The current logic
is that once A and B arrive out of order, all actionStates after the current
sequence number for the current key will be cleared.
``` java
@Override
public ActionState get(Object key, long seqNum, Action action, Event event)
throws Exception {
String stateKey = keyEncoder.generateKey(key, seqNum, action, event);
String businessKeyIdentity = keyEncoder.generateBusinessKeyIdentity(key);
boolean hasDivergence = checkDivergence(businessKeyIdentity, seqNum);
if (!actionStates.containsKey(stateKey) || hasDivergence) {
List<String> keysToRemove = new ArrayList<>();
for (String cachedKey : actionStates.keySet()) {
if (ActionStateUtil.matchesBusinessKeyIdentityWithSeqNum(
cachedKey, businessKeyIdentity, stateSeqNum ->
stateSeqNum > seqNum)) {
keysToRemove.add(cachedKey);
}
}
actionStates.keySet().removeAll(keysToRemove);
latestStateOffsets.keySet().removeAll(keysToRemove);
}
ActionState result = actionStates.get(stateKey);
return result;
}
```
So the only consequence of out-of-order arrival is increased re-run cost.
The solution I have in mind to solve this problem is to make the order
deterministic at the ingress: the processing order for a given key must be a
"pure function of the input multiset," independent of arrival interleaving.
This is achieved by merging based on the ordering information carried by the
inputs:` (originId, originSeq)`—a stable source identifier plus a monotonic
sequence number for that source for that key. After that, the meaning of
`seqNum `becomes "position after merging," still assigned when input is
accepted, still goes into checkpoint, and no changes are needed inside the
`actionstate `package. If this approach is worth discussing, I am willing to
open a discussion to talk through the details, which involves changes to
`event`.
--
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]
