wank125 opened a new pull request, #1312: URL: https://github.com/apache/flink-agents/pull/1312
Linked issue: #1308 (finding 4) ### Purpose of change **User-visible outcome.** `agentsEnv.fromDataStream(keyedStream)` without a key selector no longer keys the agent by the whole input record. Messages of the same session now share short-term memory: in the regression test the session counts 1, 2, 3 across its messages, where previously every message counted 1 because each record became its own key. This implements finding 4 of #1308 — the [DataStream input contract](https://github.com/apache/flink-agents/blob/dev/docs/content/docs/development/integrate_with_flink.md) allows passing an already-keyed stream, but the null-selector branch of `RemoteAgentBuilder.toDataStream()` replaced its key with a pass-through `x -> x` selector, and `CompileUtils` then called `keyBy` again. **Runtime flow.** `RemoteAgentBuilder.toDataStream()` now branches three ways: an explicit selector still goes to `CompileUtils.connectToAgent(DataStream, selector, plan)` (unchanged); no selector and the input is a `KeyedStream` now goes to the existing `connectToAgent(KeyedStream, plan)` path, which does not re-key (the fix); otherwise the pass-through selector is kept (unchanged — for a non-keyed input, identity is the only key available). ### Behavioral Semantics #### Key decisions - Reuse the existing `connectToAgent(KeyedStream, AgentPlan)` overload rather than trying to recover the original selector from the `KeyedStream` (it is not exposed); the keyed path preserves the key without one. - Keep identity keying for plain `DataStream` input without a selector; there is no other key to use, and all in-repo examples use this path. #### Interaction decisions | input | selector | behavior | |---|---|---| | `KeyedStream` | none | keyed by the stream's existing key (fixed; was: whole record) | | `KeyedStream` | explicit | explicit selector wins (unchanged) | | `DataStream` | none | keyed by record identity (unchanged) | | `DataStream` | explicit | keyed by the selector (unchanged) | | table input (fromTable) | n/a | converted via `toDataStream`, never a `KeyedStream` (unchanged) | #### Behavioral contracts 1. An already-keyed input connected without a selector runs the agent keyed by that stream's existing business key, so inputs sharing the key share short-term memory. 2. An explicit key selector, when given, still replaces the key. 3. A non-keyed input without a selector is still keyed by record identity. #### Failure behavior No new failure paths: `toDataStream()` still throws `IllegalStateException` before `apply`; the unchecked cast is guarded by the `instanceof` check. #### Compatibility impact A caller passing an already-keyed stream without a selector and relying on per-record identity keying would see keying change to the stream's key — that reliance contradicts the documented contract the fix restores. All other combinations are unchanged, and no in-repo caller or example relies on the old behavior. ### Tests | contract | test | |---|---| | keyed input + no selector shares memory per the existing key | `RemoteExecutionEnvironmentTest#fromDataStreamPreservesExistingKeyedStreamKeyWhenNoSelectorGiven` (new; fails on main with every same-session message counting 1) | | keyed path itself (no re-keying) | `CompileUtilsTest#testJavaKeyedStreamConnectToAgent` (existing) | | explicit selector / identity fallback | `CompileUtilsTest#testJavaNoKeyedStreamConnectToAgent` and the example-based tests (existing) | Not verified: batch-mode execution of the keyed path (the regression test runs a bounded streaming job); the Python table input path (untouched by this change). Local verification: full `runtime` module suite (1356 tests) passes on JDK 17, `spotless:check` passes. ### API No public API change. ### Documentation - [x] `doc-not-needed` (the change restores the behavior the integrate-with-Flink guide already documents) -- 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]
