You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "wenjin272 (via GitHub)" <gi...@apache.org> on 2026/10/21 03:55:04 UTC
[I] [Feature][runtime] Support durable replay for same-key inputs from multiple upstream instances [flink-agents]
wenjin272 opened a new issue, #1302:
URL: https://github.com/apache/flink-agents/issues/1302
### Search before asking
- [x] I searched the existing issues. #1301 tracks incorrect divergence detection for normal multi-Action execution; #1084 tracks event-occurrence key collisions. Neither restores the original input ordering across upstream channels.
### Description
ActionStateStore reuse currently depends on replaying each business key's inputs in the same order. The operator assigns seqNum from a checkpointed per-key counter when it starts a top-level InputEvent; seqNum is not a stable source-record identifier.
When the same business key arrives from multiple upstream parallel instances, keyBy routes those inputs to the same downstream instance but does not establish a replay-stable order across the incoming channels. A failure can therefore assign a different seqNum to the same logical input.
This is a practical limitation for agents consuming merged streams or records distributed across multiple source partitions/upstream instances under the same agent key. Correcting #1301 avoids false invalidation for normal multi-Action execution, but does not eliminate this limitation.
The [current consistency documentation](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/docs/content/docs/operations/consistency.md#conditions-for-reusing-saved-records) already requires stable per-key input order. This issue requests support for the broader topology and a precise user-facing recovery contract, rather than claiming that the existing documentation guarantees it.
### Example and impact
Take checkpoint C0 after seqNum=9 for key K. Two upstream instances then emit distinct records for K:
```text
upstream 1: A, C
upstream 2: B, D
original execution:
seq=10: A
seq=11: B
seq=12: C
seq=13: D
completed ActionStates are persisted; failure before the next checkpoint
restore C0, with a different channel interleaving:
seq=10: B
seq=11: A
seq=12: C
seq=13: D
```
Both deliveries preserve each upstream channel's order, yet the per-key merged sequence differs. The lookup for B at seq=10 cannot find B's completion saved at seq=11. Cache-miss/divergence handling can invalidate the remaining saved suffix, causing completed work to run again.
Even when C returns to its old sequence number, blindly reusing its result is unsafe if its output or Memory updates depended on the earlier processing order. For example, A and B can set the same Memory field to different values, which C then reads.
- Model/tool calls can repeat despite having saved completions.
- If order changes at the first replayed input, reuse for that key's affected replay window can be lost.
- Changes only in the interleaving of different keys do not cause this issue.
- Parallelism or keyBy alone is not sufficient to trigger it: a key with a replay-stable ordered input path can still reuse saved state.
- This is about ActionState reuse and repeated external work; it is not a claim that all Flink checkpointed state or transactional sinks lose their guarantees.
Flink's [dataflow documentation](https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/learn-flink/overview/) explains that redistribution preserves ordering within each sending/receiving subtask pair, not across all senders.
### Proposed investigation
Define how agents should recover when same-key inputs originate from multiple channels. Possible directions include durably recording the accepted per-key input order and restoring it, or introducing stable input identity together with the state-dependency semantics required for safe reuse.
A stable input ID alone is insufficient for stateful Actions: recovering a result by ID must not replay Memory changes or outputs computed against a different preceding history. Likewise, disabling divergence protection would mask the mismatch rather than solve it.
A substantial recovery-model change should receive a separate design discussion before implementation.
### Acceptance criteria
- Add a controlled recovery test with at least two upstream instances emitting distinct inputs for the same agent key, deliberately reverse their interleaving after failure, and measure actual Action/tool invocations.
- Define and implement a supported replay contract that preserves safe reuse of saved completed work for that topology. If ordering restoration is selected, verify that the accepted history is restored before applying dependent Memory updates.
- Test stateful Actions whose results depend on preceding same-key inputs, not just stateless input-to-output functions.
- Cover repeated failures and checkpoints taken while recovery is progressing; validate both Kafka and Fluss backends.
- Keep different keys isolated and retain the existing ordered-input behavior.
- Document supported topologies, ordering prerequisites, and fallback/re-execution behavior for inputs outside the supported contract. Until broader support exists, explain concretely that keyBy alone does not satisfy the ordering prerequisite.
- Do not weaken the genuine divergence safeguards tracked in #1301.
### Version and validation status
Flink Agents 0.4-SNAPSHOT; source analysis used audit baseline 47daff57b5c4189c80dd7037660f2a7832270879, including OperatorStateManager.initOrIncSequenceNumber and the Kafka/Fluss lookup paths.
The sequence-number dependency and invalidation paths were inspected. The separate #1301 prototype exercised controlled changed-input histories through real operator snapshot/restore with mocked broker transport. A live multi-upstream network-shuffle failover reproduction for this issue has not yet been run.
### Are you willing to submit a PR?
- [ ] I'm willing to submit a PR!
--
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: issues-unsubscribe@flink.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Feature][runtime] Support durable replay for same-key inputs from multiple upstream instances [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on issue #1302:
URL: https://github.com/apache/flink-agents/issues/1302#issuecomment-6491010794
Thanks for the clarification @da-daken. The C example was meant to explain why retaining C’s saved result would be unsafe once its preceding history changes. I agree that the current suffix invalidation protects against this, at the cost of re-execution.
Regarding the proposed direction, I have a couple of concerns:
- How would `originId` and `originSeq` be defined and remain stable across recovery and rescaling? If they are tied to upstream subtask identities or arrival order, the same logical input could receive different values after recovery.
- For an unbounded stream, how would the merge decide when it is safe to process the next input? Ordering metadata alone does not tell us whether an earlier input is still on its way from another source. Waiting could block progress indefinitely, while skipping an idle source based on a timeout could make the order differ across replays.
The main purpose of this issue is to record the known limitation, which is why I marked it as minor. I’d prefer to revisit the solution after ActionState has been further hardened through production use. Its recovery model may evolve during that process, potentially changing the assumptions behind a design proposed today. For now, I think we can keep this issue open to track the limitation without committing to a particular approach.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Feature][runtime] Support durable replay for same-key inputs from multiple upstream instances [flink-agents]
Posted by "da-daken (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org