You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "wank125 (via GitHub)" <gi...@apache.org> on 2026/10/22 01:27:20 UTC
[PR] [runtime][java] Preserve an existing KeyedStream's business key when no key selector is given [flink-agents]
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][java] Preserve an existing KeyedStream's business key when no key selector is given [flink-agents]
Posted by "wank125 (via GitHub)" <gi...@apache.org>.
wank125 commented on PR #1312:
URL: https://github.com/apache/flink-agents/pull/1312#issuecomment-6355147722
The single remaining red check (`it-python [flink-2.3]`) looks like a model-behavior flake rather than a code issue: `react_agent_test.py::test_react_agent_no_output_schema_on_remote_runner` asserts that the LLM calls the `multiply` tool, but across all three attempts (pytest-rerunfailures) the model answered the arithmetic directly without invoking any tool ("actual invocations: []"). The same test and code passed on the same flink-2.3 combo in the previous CI round; only the randomly generated inputs differ. Nothing in this PR affects that path — all Python e2e tests already pass explicit key selectors. A re-run should settle it.
--
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: [PR] [runtime][java] Preserve an existing KeyedStream's business key when no key selector is given [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on PR #1312:
URL: https://github.com/apache/flink-agents/pull/1312#issuecomment-6379268373
Thanks again for the fix, @wank125! With the 0.4 release process starting on Monday, I pushed two follow-up commits to this branch to address the remaining review feedback and keep release preparation moving. Both commits include you as a co-author.
The changes are:
- Require an explicit key selector for Table input in both Java and Python, remove Java's identity-key fallback, and add Table contract tests.
- Align the Java review examples around product IDs: deserialize JSON into `ProductReview` upstream for the single-agent/YAML examples, pass `ProductReviewSummary` directly from the multi-agent window to the suggestion agent, and key the ReAct Row input by `id`.
- Generate valid JSON when constructing prompts, and synchronize the four Quickstart pages, integration guide, and API documentation.
Validation passed: 34 focused Java tests, 35 Python remote-environment tests, formatting and documentation checks, plus live runs of all four Java examples against local Ollama. The multi-agent example's existing final partial-window limitation with bounded input remains unchanged.
The follow-up commits are `bbb61b30f` and `b3276f9c0`. Please take a look when you have a chance—feedback and further adjustments are very welcome!
Generated-by: Codex CLI 0.153.4 (GPT-6)
--
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: [PR] [runtime][java] Preserve an existing KeyedStream's business key when no key selector is given [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 merged PR #1312:
URL: https://github.com/apache/flink-agents/pull/1312
--
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: [PR] [runtime][java] Preserve an existing KeyedStream's business key when no key selector is given [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on code in PR #1312:
URL: https://github.com/apache/flink-agents/pull/1312#discussion_r4355491346
##########
examples/src/main/java/org/apache/flink/agents/examples/WorkflowSingleAgentExample.java:
##########
@@ -80,7 +80,7 @@ public static void main(String[] args) throws Exception {
// Use the ReviewAnalysisAgent to analyze each product review.
DataStream<Object> reviewAnalysisResStream =
agentsEnv
- .fromDataStream(productReviewStream)
+ .fromDataStream(productReviewStream, value -> value)
Review Comment:
Could we also align the Java examples with the Python examples by using typed records and product IDs as keys, instead of adding `value -> value` everywhere?
For the single-agent and YAML examples, deserialize JSON into `ProductReview` upstream:
```java
DataStream<ProductReview> productReviewStream =
rawStream
.map(json -> MAPPER.readValue(json, ProductReview.class))
.returns(ProductReview.class);
DataStream<Object> output =
agentsEnv
.fromDataStream(productReviewStream, ProductReview::getId)
.apply(new ReviewAnalysisAgent())
.toDataStream();
```
The YAML example can use the same typed input and selector while keeping `.apply("review_analysis_agent")`. `ReviewAnalysisAgent.processInput` would consume the object directly:
```java
ProductReview review =
(ProductReview) InputEvent.fromEvent(event).getInput();
ctx.getShortTermMemory().set("id", review.getId());
// Build the prompt from review.getId() and review.getReview().
```
This keeps JSON parsing at the input boundary and avoids parsing again inside the selector or agent.
For the multi-agent example, let the window emit `ProductReviewSummary` directly:
```java
// Change the window function's output type to ProductReviewSummary:
// ProcessWindowFunction<
// ProductReviewAnalysisRes, ProductReviewSummary, String, TimeWindow>
ProductReviewSummary summary =
new ProductReviewSummary(key, formattedPercentages, reasonList);
out.collect(summary); // Instead of MAPPER.writeValueAsString(summary).
```
Then key the typed stream by product ID:
```java
DataStream<ProductReviewSummary> summaries =
reviewAnalysisResStream
.keyBy(ProductReviewAnalysisRes::getId)
.window(TumblingProcessingTimeWindows.of(Duration.ofMinutes(1)))
.process(new AggregateScoreDistributionAndDislikeReasons());
DataStream<Object> suggestions =
agentsEnv
.fromDataStream(summaries, ProductReviewSummary::getId)
.apply(new ProductSuggestionAgent())
.toDataStream();
```
`ProductSuggestionAgent` can consume the object and serialize it when constructing the prompt:
```java
ProductReviewSummary summary =
(ProductReviewSummary) InputEvent.fromEvent(event).getInput();
String promptInput = MAPPER.writeValueAsString(summary);
```
The ReAct example already produces `Row` values, so it only needs:
```java
.fromDataStream(productReviewStream, row -> (String) row.getField("id"))
```
Please update the affected callers and existing tests for these input-type changes, and synchronize the Java snippets in the single-agent, multi-agent, ReAct, and YAML Quickstart pages. Those snippets still use `fromDataStream(input)` without a selector and will fail under the new validation.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/env/RemoteExecutionEnvironment.java:
##########
@@ -229,10 +237,25 @@ public DataStream<Object> toDataStream() {
if (keySelector != null) {
outputDataStream =
CompileUtils.connectToAgent(inputDataStream, keySelector, agentPlan);
- } else {
- // If no key selector provided, use a simple pass-through key selector
+ } else if (inputDataStream instanceof KeyedStream) {
+ // An already-keyed input carries its business key; connect through the
+ // keyed-stream path to preserve that key, instead of replacing it with a
+ // pass-through selector that would split a session across its messages.
+ outputDataStream =
+ CompileUtils.connectToAgent(
+ (KeyedStream<T, ?>) inputDataStream, agentPlan);
+ } else if (allowIdentityKeyFallback) {
Review Comment:
I’d prefer to align the Table path too: require an explicit selector in both Java and Python. Could we remove `allowIdentityKeyFallback`, reject Java Table input without a selector, and update the docs? The “without keying” Javadoc is misleading, since identity keying still determines memory sharing.
--
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: [PR] [runtime][java] Preserve an existing KeyedStream's business key when no key selector is given [flink-agents]
Posted by "wank125 (via GitHub)" <gi...@apache.org>.
wank125 commented on PR #1312:
URL: https://github.com/apache/flink-agents/pull/1312#issuecomment-6342820926
Thanks for the direction @wenjin272 — addressed in f6fbe45b on top of fe8d66e7. Key selection now follows the same three rules in both languages:
- An explicit selector always wins, even over the key of an already-keyed input (this also fixes Python silently ignoring an explicit selector on keyed input).
- Otherwise a `KeyedStream` input keeps its existing key.
- If neither is available, the pipeline fails with a clear error (`IllegalArgumentException` in Java, `RuntimeError` in Python) instead of silently keying by the whole record.
Coverage: all four input/selector combinations in both languages — `RemoteExecutionEnvironmentTest` in Java (including a job-level regression test asserting shared short-term memory per session key and isolation across sessions) and `test_key_selection_combinations` in Python. All affected Java examples now pass an explicit selector, and `integrate_with_flink.md` plus the `fromDataStream`/`from_datastream` docs state the rules.
Compatibility note: an unkeyed input without a selector changes from silent whole-record keying to a clear error, as requested. One judgement call to confirm: Java's `fromTable(input)` documents a "without keying" mode and converts the table to an unkeyed `DataStream`, so I kept identity keying for that path only (`allowIdentityKeyFallback`) — Python's `from_table` already requires a selector. Happy to extend the same error to the table path if you prefer full alignment.
Full `runtime` + `examples` suites pass locally after rebasing onto the latest main (efd7b8f3).
--
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