sangkyoonnam opened a new issue, #1335: URL: https://github.com/apache/flink-agents/issues/1335
### Search before asking - [x] I searched in the [issues](https://github.com/apache/flink-agents/issues) and found nothing similar. ### Description The agent operator (`ActionExecutionOperator`) holds the Flink-managed sensory and short-term memory, pending actions and input events, and, when an Action State Store is configured, its recovery markers. `CompileUtils.connectToAgent` creates it with `transform(...)` and `setParallelism(...)` but no UID, so Flink derives its operator ID from the stream topology. Some upstream topology changes can change that ID, and the agent's saved state then no longer maps back to the operator on restore. Neither `AgentBuilder` exposes that UID. Java's raw `toDataStream()` output needs a cast or the underlying transformation, and the typed Java outputs and all Python `to_datastream()` outputs add a conversion operator after the agent, so `.uid()` on them sets the UID of that operator. For snapshots of agent operators without an explicit UID, `KafkaActionStateCleanupTool plan` needs `--operator-uid-hash`, as in the consistency docs. wenjin272 agreed in #1327 to expose this before 1.0. My proposal: - Java `AgentBuilder uid(String uid)` and Python `uid(self, uid: str) -> "AgentBuilder"`, called before the first output conversion, since both builders cache the agent stream. The builders pass the UID to `connectToAgent`, which sets it on the agent transformation. This covers `DataStream` and `Table` inputs, and agents loaded from YAML and selected with `apply(name)`. The UID identifies this agent operator in the job, so it must be unique there, and a blank UID is rejected. - When no UID is given, the graph is built exactly as today, so generated IDs don't change. Setting a UID when redeploying an existing job does change it, so the docs should recommend setting it from the first deployment, and point to a migration for existing savepoints, such as remapping the operator identifier with Flink's State Processor API. `--allowNonRestoredState` skips restoring unmatched agent state; it doesn't migrate it. - Tests for propagation from both languages, including the `Table` and YAML paths, unchanged generated IDs when the UID is unset, and a stable agent operator ID across an upstream topology change when it's set. A docs update in the operations section with that guidance and a cleanup example using `--operator-uid`. I'll go with a builder method unless you'd rather have it as a config option or an argument to `apply`. ### Are you willing to submit a PR? - [x] 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
