joeyutong opened a new issue, #1308: URL: https://github.com/apache/flink-agents/issues/1308
### Description Follow-up to #1283 for the **Overview & Setup, Quickstart, and Integrate with Flink** walkthrough. This issue tracks five findings: - [ ] 1. Setup: make the Apple Silicon build-dependency workaround work across pip versions. - [ ] 2. Setup: keep the working directory consistent between the source build and distribution JAR copy. - [ ] 3. Multi-Agent Quickstart: prevent the bounded file source from ending the job before processing-time windows emit. - [ ] 4. Integrate with Flink: preserve an existing Java KeyedStream's business key. - [ ] 5. Integrate with Flink: adapt Python scalar output to a single-column Table Row. **Findings 4 and 5 are independent implementation bugs discovered while checking the Integrate guide. The documented contracts are reasonable; these are follow-up runtime problems, rather than defects in the documentation itself.** They are included here to retain the walkthrough context and can be fixed separately. The links below are pinned to the audited source baseline, [`47daff57`](https://github.com/apache/flink-agents/commit/47daff57b5c4189c80dd7037660f2a7832270879). ### 1. Setup: Apple Silicon workaround no longer constrains isolated builds on pip 26.2+ **Documentation:** [Setup → From Official Release, Apple Silicon warning](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/docs/content/docs/get-started/setup.md#L180-L188). The warning recommends: ```sh echo "setuptools<82" > /tmp/constraint.txt PIP_CONSTRAINT=/tmp/constraint.txt pip install flink-agents apache-flink==${FLINK_VERSION} ``` **Problem:** On the source-build path described in the warning, the build needs a setuptools version that still provides `pkg_resources`. Starting with pip 26.2, `PIP_CONSTRAINT` no longer applies to isolated build dependencies. The command can therefore select setuptools 82+ and fail again with `ModuleNotFoundError: No module named 'pkg_resources'`. **Verification:** An offline minimal PEP 517 build probe, using real setuptools wheels and a backend that imports `pkg_resources`, reproduced the distinction with actual pip releases. With pip 26.2, `PIP_CONSTRAINT` alone selected setuptools 82.0.1 and failed; adding `PIP_BUILD_CONSTRAINT` selected setuptools 75.3.4 and succeeded. Setting both variables to the same file also succeeded with pip 25.2, 25.3, and 26.1.2. This verified the isolated-build mechanism, rather than a complete Beam installation. **Suggested fix:** Use one command that covers the tested old and new pip versions: ```sh echo "setuptools<82" > /tmp/constraint.txt PIP_CONSTRAINT=/tmp/constraint.txt \ PIP_BUILD_CONSTRAINT=/tmp/constraint.txt \ pip install flink-agents "apache-flink==${FLINK_VERSION}" ``` `PIP_BUILD_CONSTRAINT` was added in pip 25.3, so replacing the old variable outright would leave older pip versions uncovered. Both variables can point to the same file. See the [pip build-constraints guide](https://pip.pypa.io/en/stable/user_guide/#build-constraints) and [pip 26.2 changelog](https://pip.pypa.io/en/stable/news/#v26-2). ### 2. Setup: the source-build instructions leave the reader outside the repository **Documentation:** [Setup → From Source, build commands](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/docs/content/docs/get-started/setup.md#L199-L209), followed by [copying the source-built distribution JAR](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/docs/content/docs/get-started/setup.md#L319-L328). **How to reproduce:** Follow the commands from the parent of a freshly cloned repository: ```sh cd flink-agents ./tools/build.sh cd .. # Later in the same page: cp dist/flink-${FLINK_VERSION%.*}/target/flink-agents-dist-flink-${FLINK_VERSION%.*}-${FLINK_AGENTS_VERSION}.jar $FLINK_HOME/lib/ ``` **Expected:** The copy finds the distribution JAR that was just built. **Actual:** `dist/...` is resolved relative to the clone's parent directory, so the copy fails even when the JAR exists under the repository. An isolated path fixture with a pre-existing placeholder JAR confirmed that leaving the repository makes the copy fail, and returning to its root makes it succeed. The fixture did not rebuild the project. **Suggested fix:** Remove the trailing `cd ..` so subsequent repository-relative commands run from `flink-agents/`. ### 3. Multi-Agent Quickstart: bounded input can end before the one-minute window emits **Documentation:** [file sources in both languages](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/docs/content/docs/get-started/quickstart/create-agent/multiple-agents.md#L110-L151), [Python processing-time window](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/docs/content/docs/get-started/quickstart/create-agent/multiple-agents.md#L317-L321), [Java processing-time window](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/docs/content/docs/get-started/quickstart/create-agent/multiple-agents.md#L378-L382), and [expected suggestion output](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/docs/content/docs/get-started/quickstart/create-agent/multiple-agents.md#L542-L544). The same composition appears in [the Python example](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/python/flink_agents/examples/quickstart/workflow_multiple_agent_example.py#L125-L173) and [the Java example](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/examples/src/main/java/org/apache/flink/agents/examples/WorkflowMultipleAgentExample.java#L154-L194). **Problem:** The filesystem source has no `source.monitor-interval`, so it is bounded. Both examples then aggregate analysis results in one-minute tumbling processing-time windows before sending them to the suggestion Agent. If all inputs finish before a pending timer fires, the job can finish without emitting that window. Fast analysis can yield no suggestions at all; otherwise the last window can still be lost. The one-minute duration also follows clock-aligned window boundaries, rather than starting a fresh minute with the first record. **Verification:** A real Flink 2.2.1 MiniCluster probe reproduced the bounded-source/window composition without Agents or model calls. Three input records reached a direct sink, but the window branch emitted zero records. At the first input, 15,246 ms remained until the processing-time timer; the complete job finished in 1,493 ms. This demonstrates the source/window termination problem; the complete Ollama Quickstart was not run. **Suggested minimal fix:** Keep the intended streaming example running by adding the following option to the source in **both runnable examples and both documentation snippets**: ```java .option("source.monitor-interval", "1 s") ``` This is the file-discovery interval; the aggregation window stays one minute. Explain that the example keeps running and should be cancelled after observing suggestions. See [Flink filesystem source monitoring](https://nightlies.apache.org/flink/flink-docs-release-2.2/docs/connectors/table/filesystem/#directory-watching). Verify that the updated examples produce suggestions before cancellation. ### 4. Integrate follow-up implementation bug: Java replaces an existing KeyedStream's key **Documentation context:** [Integrate with Flink → DataStream input contract](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/docs/content/docs/development/integrate_with_flink.md#L122). The guide allows an existing KeyedStream instead of supplying a KeySelector. **How to reproduce:** In a STREAMING job with parallelism 1, key records by session ID before passing them to the Agent: ```java KeySelector<String, String> sessionKey = value -> value.split(":")[0]; KeyedStream<String, String> keyed = input.keyBy(sessionKey); agents.fromDataStream(keyed).apply(new CountingAgent()).toDataStream(); ``` Use a model-free CountingAgent that reads `count` from `ctx.getShortTermMemory()`, increments it, stores it, and emits the stored value for each input. Disable Memory TTL for the probe. Compare with `agents.fromDataStream(keyed, sessionKey)` using the same Agent and input. **Actual real-job results for one session:** | Input | Existing KeyedStream, no selector | Explicit session selector control | | --- | --- | --- | | `same:first` | 1 | 1 | | `same:second` | **1** | **2** | | `same:first` again | 2 | 3 | A second, interleaved session stayed independent in the control. Both branches ran in the same completed Flink job. Sink output and Trace Log business keys/short-term writes agreed: the failing path used `same:first` and `same:second` as separate keys. Repeating the identical message reused its Memory, which rules out Memory simply being cleared after every input. **Cause:** The [null-selector branch](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/runtime/src/main/java/org/apache/flink/agents/runtime/env/RemoteExecutionEnvironment.java#L228-L235) passes `x -> x`; [CompileUtils](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/runtime/src/main/java/org/apache/flink/agents/runtime/CompileUtils.java#L71-L79) calls `keyBy` again. The original business key is replaced with the whole input value, splitting a conversation's Memory across its messages. **Suggested fix:** When no new selector is provided and the input is already a KeyedStream, connect it through the existing keyed-stream path. Keep explicit-selector handling separate. Add a runtime regression test asserting shared Memory for different messages in the same session and isolation between different sessions. Preserve the documentation contract; this is an implementation fix. ### 5. Integrate follow-up implementation bug: Python scalar output is not converted to a single-column Row **Documentation context:** [the Python single-column Schema example](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/docs/content/docs/development/integrate_with_flink.md#L181-L188) and [the scalar-to-single-column contract](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/docs/content/docs/development/integrate_with_flink.md#L287-L289). **How to reproduce:** Use an importable, model-free Python Agent whose input action sends `OutputEvent(output=42)`, and execute the resulting Table: ```python schema = Schema.new_builder().column("result", DataTypes.INT()).build() output_table = ( agents_env.from_table(input=input_table, key_selector=lambda row: row.id) .apply(ScalarAgent()) .to_table(schema=schema) ) with output_table.execute().collect() as rows: print(list(rows)) ``` An insert-only input such as `t_env.from_elements([(1, "hello")], ["id", "input"])` is sufficient. **Expected:** One row containing `42`. **Actual:** The action executes and emits the integer, but the job fails in PyFlink's Row coder: ```text AttributeError: 'int' object has no attribute 'get_fields_by_names' ``` **Verification:** Six real Python/Flink jobs confirmed the boundary: | Input API | Agent output | Terminal | Result | | --- | --- | --- | --- | | `from_datastream` | `42` | single-column `to_table` | FAILED, error above | | `from_table` | `42` | single-column `to_table` | FAILED, same error | | `from_datastream` | `{"result": 42}` | single-column `to_table` | FINISHED, row containing 42 | | `from_table` | `{"result": 42}` | single-column `to_table` | FINISHED, row containing 42 | | `from_table` | `Row(result=42)` | single-column `to_table` | FINISHED, row containing 42 | | `from_datastream` | `42` | `to_datastream(output_type=Types.INT())` | collected `[42]` successfully | **Cause:** [`_row_stream`](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/python/flink_agents/runtime/remote_execution_environment.py#L202-L214) declares Row output, but [`to_row`](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/python/flink_agents/runtime/output_type_utils.py#L263-L293) returns a scalar unchanged. The Row coder therefore receives an integer. The [Java conversion](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/runtime/src/main/java/org/apache/flink/agents/runtime/OutputTypeUtils.java#L169-L174) already wraps the corresponding value in a Row. **Suggested fix:** At the top-level Table/Row adaptation boundary, wrap scalar output in a Row when the declared row type has exactly one physical column. Preserve nested scalar fields and map/array column semantics. Add runtime coverage for both input APIs and successful dict/Row controls. Keep the documented scalar contract and align Python behavior with Java; this is an implementation fix. ### Version and verification scope - Flink Agents `0.4-SNAPSHOT`, source baseline `47daff57b5c4189c80dd7037660f2a7832270879`; JAR/wheel built from that snapshot, with source and packaged artifacts checked for consistency. - Actual integration reproductions: Flink/PyFlink 2.2.1, JDK 21.0.10, Python 3.11.15, local MiniCluster, STREAMING, parallelism 1. No model calls or mocked Agent execution. - The window finding was reproduced as an isolated Flink source/window job. The pip finding used actual pip releases and a minimal isolated-build probe. The working-directory finding used a path fixture. - Complete model-backed Quickstart runs remain unverified because no Ollama service/model was available. This issue does not claim that those end-to-end runs passed. -- 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]
