From 469efd68ce8420c50a675bfc8adb3b56a6d32b70 Mon Sep 17 00:00:00 2001 From: joeyutong Date: Tue, 20 Oct 2026 10:17:25 +0800 Subject: [PATCH 01/13] [docs] Improve consistency and configuration guidance Add an operations guide for consistency setup, recovery, retention, cleanup, and migration, and organize the configuration reference by user task. Consolidate shared recovery guidance and retain feature-specific contracts. Related to #1234. Generated-by: Codex CLI 0.163.0-alpha.10.2 (GPT-6) Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 1153/1153 AI-Contributed/UT: 0/0 --- docs/content/_index.md | 2 +- .../docs/development/calling-llms/chat.md | 12 +- .../development/calling-llms/model_routing.md | 12 +- .../docs/development/calling-llms/tool_use.md | 10 +- .../docs/development/create-agent/actions.md | 62 ++- .../docs/development/decision_models.md | 6 +- .../docs/development/memory/memory_events.md | 2 +- .../memory/sensory_and_short_term_memory.md | 4 +- .../docs/development/multi-agent/subagents.md | 48 +- docs/content/docs/get-started/overview.md | 2 +- docs/content/docs/operations/configuration.md | 425 +++++++++--------- docs/content/docs/operations/consistency.md | 384 ++++++++++++++++ docs/content/docs/operations/deployment.md | 61 +-- docs/content/docs/operations/monitoring.md | 123 ++++- 14 files changed, 806 insertions(+), 347 deletions(-) create mode 100644 docs/content/docs/operations/consistency.md diff --git a/docs/content/_index.md b/docs/content/_index.md index 06d239970..5e8d316a1 100644 --- a/docs/content/_index.md +++ b/docs/content/_index.md @@ -45,7 +45,7 @@ It provides a step by step introduction on how to install Flink Agents and build ### How to Operate an Agent -* [Deployment]({{< ref "docs/operations/deployment" >}}), [Configuration]({{< ref "docs/operations/configuration" >}}), and [Monitoring]({{< ref "docs/operations/monitoring" >}}) - Deploy, configure, and monitor agents +* [Deployment]({{< ref "docs/operations/deployment" >}}), [Consistency]({{< ref "docs/operations/consistency" >}}), [Configuration]({{< ref "docs/operations/configuration" >}}), and [Monitoring]({{< ref "docs/operations/monitoring" >}}) - Deploy, configure, recover, and monitor agents <---> diff --git a/docs/content/docs/development/calling-llms/chat.md b/docs/content/docs/development/calling-llms/chat.md index e86388fff..6d5ca28b3 100644 --- a/docs/content/docs/development/calling-llms/chat.md +++ b/docs/content/docs/development/calling-llms/chat.md @@ -215,7 +215,7 @@ support `gather`. For general await behavior and durable execution, see [Async Execution]({{< ref "docs/development/create-agent/actions#async-execution" >}}) and [Durable Execution]({{< ref "docs/development/create-agent/actions#durable-execution" >}}). -With durable execution enabled, chat calls reuse recorded responses by Action +With an Action State Store configured, chat calls match saved responses by Action execution and await order, not by model or message contents. Existing model and tool timeouts apply; `ctx.chat` adds no whole-call timeout. @@ -259,13 +259,13 @@ the failure. Model-call failures are reported after configured retries and routing fallbacks -are exhausted. See [Core Options]({{< ref "docs/operations/configuration#core-options" >}}) +are exhausted. See [Model calls, tools, and retries]({{< ref "docs/operations/configuration#model-calls-tools-and-retries" >}}) for retry settings and [Model Routing]({{< ref "docs/development/calling-llms/model_routing" >}}) -for fallbacks. +for fallbacks. The terminal `ChatResponseEvent` also reports the retry count and +total retry wait in seconds. -For how calls behave after a failure and recovery, see -[Recovery Behavior]({{< ref "docs/development/create-agent/actions#recovery-behavior" >}}) -in Actions. +For how calls behave during recovery, see +[Recovery flow]({{< ref "docs/operations/consistency#recovery-flow" >}}). ## Multimodal Messages {#message-content-blocks} diff --git a/docs/content/docs/development/calling-llms/model_routing.md b/docs/content/docs/development/calling-llms/model_routing.md index 00be59daa..6f95e89bc 100644 --- a/docs/content/docs/development/calling-llms/model_routing.md +++ b/docs/content/docs/development/calling-llms/model_routing.md @@ -281,7 +281,7 @@ behaviors: For candidates `[small, medium, big]`, selecting `big` with fallback enabled tries `big`, then `small`, then `medium`. Model-call retries run before fallback; see -[Core Options]({{< ref "docs/operations/configuration#core-options" >}}). +[Model calls, tools, and retries]({{< ref "docs/operations/configuration#model-calls-tools-and-retries" >}}). Failures before the call, such as an unresolved model resource, can proceed directly to the next candidate without retries. @@ -351,11 +351,11 @@ metadata when distinguishing their usage. ## Recovery -With durable execution enabled, completed routing decisions and model calls can -be reused after recovery. Without it, a request may be routed again and select a -different model. See -[Recovery Behavior]({{< ref "docs/development/create-agent/actions#recovery-behavior" >}}) -for the persistence and replay guarantees. +With an Action State Store configured, recovery can reuse saved routing decisions +and model call outcomes. A request whose decision was not saved may be routed +again and select a different model. See +[Recovery flow]({{< ref "docs/operations/consistency#recovery-flow" >}}) +for when saved work can be reused. ## Examples diff --git a/docs/content/docs/development/calling-llms/tool_use.md b/docs/content/docs/development/calling-llms/tool_use.md index ce33ef845..10e9c6050 100644 --- a/docs/content/docs/development/calling-llms/tool_use.md +++ b/docs/content/docs/development/calling-llms/tool_use.md @@ -435,12 +435,10 @@ the request has a single tool call. Configure timeouts for blocking operations inside each tool, such as an HTTP client read timeout. A batch timeout does not undo external effects or guarantee that an already -running tool stops. Tools that change external systems should be idempotent, -because a call may run again after recovery. See -[Configuration]({{< ref "docs/operations/configuration" >}}) for the full async batch -and execution-capacity options, and -[Durable Execution]({{< ref "docs/development/create-agent/actions#durable-execution" >}}) -for recovery guarantees. +running tool stops. For recovery guarantees and protection against repeated effects, +see [Consistency]({{< ref "docs/operations/consistency#limits-on-external-side-effects" >}}). +See [Configuration]({{< ref "docs/operations/configuration" >}}) for the full async batch +and execution-capacity options. ## Tool response content and metadata diff --git a/docs/content/docs/development/create-agent/actions.md b/docs/content/docs/development/create-agent/actions.md index 097ae9e92..6ba4e2971 100644 --- a/docs/content/docs/development/create-agent/actions.md +++ b/docs/content/docs/development/create-agent/actions.md @@ -562,8 +562,9 @@ For example, if an Action fails after a profile lookup has completed and its res been persisted, recovery can reuse that profile instead of querying the service again. Code outside the durable call may still run again when the Action is re-executed. -Durable execution requires an external action state store. Configure it as described in -[Exactly-Once Action Consistency]({{< ref "docs/operations/deployment#exactly-once-action-consistency" >}}). +To save and reuse call outcomes during recovery, configure checkpointing and an external +Action State Store as described in +[Consistency]({{< ref "docs/operations/consistency#configure-an-action-state-store" >}}). #### A Durable Call @@ -628,18 +629,10 @@ Java's `getResultClass()` provides the type needed to deserialize a saved result #### Recovery Behavior -When recovery re-executes an unfinished Action and reaches a durable call: - -| State of the matching call | Recovery behavior | -| --- | --- | -| A successful result has been persisted | Return the saved result without calling the function again. | -| A failure has been persisted | Replay the failure without calling the function again. | -| No completed outcome has been persisted | Execute the function again, or use the reconciler for a previously recorded in-flight call. | - -An external operation may finish before its result is persisted. If a failure occurs in -that interval, calling the operation again can repeat its side effects. For such -operations, use the external service's idempotency support or a reconciler that can -recover the existing outcome. +See [Recovery flow]({{< ref "docs/operations/consistency#recovery-flow" >}}) for how recovery +handles saved outcomes and in-flight calls, and +[Limits on external side effects]({{< ref "docs/operations/consistency#limits-on-external-side-effects" >}}) +for operations that may repeat. #### Reconciling an In-flight Call @@ -799,15 +792,31 @@ worker and batch limits. Concurrency between Actions has additional requirements the cross-language Flink versions below. - **Python Actions:** use `async def` and the framework's futures and `gather` APIs. -See [Core Options]({{< ref "docs/operations/configuration#core-options" >}}) for worker-pool -and concurrency limits. These scheduling requirements apply to both ordinary and durable -async calls. Durable calls additionally require the configured action state store. +The experimental parallel engine also uses `num-async-threads` to limit concurrent +Action workers. `max-in-flight-input-records` limits admitted inputs per operator +subtask, including inputs queued for a key already being processed. At that limit, +further admission waits for a record to finish. Inputs for the same key commit in +input order. The continuation engine and plans with Python Actions ignore this +input limit. If the experimental engine behaves unexpectedly, set +`parallel-execution.enabled=false` and report an issue. + +See [Runtime and async execution]({{< ref "docs/operations/configuration#runtime-and-async-execution" >}}) +for worker-pool and batch limits. These scheduling requirements apply to both +ordinary and durable async calls. To persist durable call outcomes for recovery, +also configure an [Action State Store]({{< ref "docs/operations/consistency#configure-an-action-state-store" >}}). #### Cross-language Async Requirements Async calls between Java and Python require Flink **1.20.5+**, **2.0.2+**, **2.1.3+**, -**2.2.1+**, or **2.3+**, within the respective release line. This includes built-in chat -and context-retrieval Actions that use a Resource implemented in the other language. +**2.2.1+**, or **2.3+**, within the respective release line. This includes built-in Chat +and context retrieval Actions that use a Resource implemented in the other language. + +The built-in Chat and context retrieval Actions fall back to synchronous calls if +the Resource is implemented in the other language and the runtime cannot support +that async call. Java Decision Model calls that reach Python use the same fallback. +Setting `chat.async`, `decision.async`, or `rag.async` to `false` also selects +synchronous execution. These switches do not disable durable execution; saving +call outcomes still requires an Action State Store. A call may cross into the other language and call back once into its starting language. For example, a Python Action can call a Java chat model setup that uses a Python connection. @@ -884,10 +893,17 @@ missing value. #### Condition Evaluation Failures -Reading a missing attribute without a guard or applying an operation to an incompatible -value can fail while evaluating a condition for an Event. For failure handling options, see -[`action.trigger-condition.evaluate-failure-strategy`]({{< ref "docs/operations/configuration#core-options" >}}) -in Configuration. +Reading a missing attribute without a guard, applying an operation to an +incompatible value, or returning a non-Boolean value can fail while evaluating a +condition for an Event. The default `WARN_AND_SKIP` strategy logs a warning and +treats that condition as false; later OR conditions can still match the Action. +`FAIL` throws `IllegalStateException` and fails the Flink task, whose recovery +follows the job's restart configuration. + +Plan-validation failures and failures when compiling or statically checking a +condition occur during initialization and are not handled by this strategy. +See [`action.trigger-condition.evaluate-failure-strategy`]({{< ref "docs/operations/configuration#condition-handling" >}}) +for the configuration option. #### Limitations diff --git a/docs/content/docs/development/decision_models.md b/docs/content/docs/development/decision_models.md index 2848fb29f..fb3506272 100644 --- a/docs/content/docs/development/decision_models.md +++ b/docs/content/docs/development/decision_models.md @@ -534,8 +534,10 @@ when the runtime does not support asynchronous Python calls. When an Action State Store is configured, successful and failed provider outcomes are recorded through the durable execution API, so recovery can replay persisted -outcomes. A provider request whose outcome was not persisted may run again after -recovery. Each attempt emits model execution reports. Retry metrics and available +outcomes. See [Consistency]({{< ref "docs/operations/consistency#recovery-flow" >}}) +for recovery guarantees. + +Each attempt emits model execution reports. Retry metrics and available token counts are recorded under the named model resource; missing token counts remain unreported. Accumulated retry metrics are also recorded when execution exits with an exception. diff --git a/docs/content/docs/development/memory/memory_events.md b/docs/content/docs/development/memory/memory_events.md index fd5bc9c14..49b08493e 100644 --- a/docs/content/docs/development/memory/memory_events.md +++ b/docs/content/docs/development/memory/memory_events.md @@ -238,7 +238,7 @@ Because `agent-run.begin-event` is disabled by default, the Trace Log has no bui ## Semantics & Caveats -- **Trace Log level.** Configure [recording targets]({{< ref "docs/operations/configuration#trace-log-targets" >}}) with `scope: EVENT_ONLY` and `level: VERBOSE` when full memory-event attributes are required, or select individual Event names with an entity scope. Memory-event generation and log selection are independent: the emission switches determine whether Events are produced, while targets determine whether they are written to Trace Log. +- **Trace Log level.** Configure [recording targets]({{< ref "docs/operations/monitoring#recording-targets" >}}) with `scope: EVENT_ONLY` and `level: VERBOSE` when full memory-event attributes are required, or select individual Event names with an entity scope. Memory-event generation and log selection are independent: the emission switches determine whether Events are produced, while targets determine whether they are written to Trace Log. - **Durable execution.** A long-term memory operation wrapped in `durable_execute` is not recorded again when the durable result is replayed from state. The generated events describe operations that actually executed in the current attempt. - **Observation failures.** Unsupported short-term or sensory values are omitted. Invalid long-term records are skipped individually after a valid batch is parsed, but a bridge, batch-serialization, or malformed-payload failure drops the current partition key's LTM observation batch. Warnings do not include observed values. These failures do not fail the action or roll back a completed long-term backend operation. - **Failed actions.** A failed action does not reach the memory-event emission boundary. The operator task fails and recovery rebuilds the interpreter, so queued actions do not continue in the same interpreter instance after the failure. diff --git a/docs/content/docs/development/memory/sensory_and_short_term_memory.md b/docs/content/docs/development/memory/sensory_and_short_term_memory.md index a3e3e9ed4..d3576d7a3 100644 --- a/docs/content/docs/development/memory/sensory_and_short_term_memory.md +++ b/docs/content/docs/development/memory/sensory_and_short_term_memory.md @@ -290,7 +290,9 @@ Sensory Memory is automatically cleared by the framework after each agent run co During execution, sensory memory data is checkpointed by Flink for fault tolerance. However, once the run completes, all sensory memory is cleared and will not be available in subsequent runs. {{< /hint >}} -### Short-Term Memory + + +### Short-Term Memory {#short-term-memory-ttl} Short-term memory can be configured with a time-to-live (TTL) so that older state expires automatically. This is useful for agents that may run for a long time: if the agent only needs recent memories, expiring historical data directly keeps the stored state focused on the latest context and avoids retaining stale information. diff --git a/docs/content/docs/development/multi-agent/subagents.md b/docs/content/docs/development/multi-agent/subagents.md index cbd696f6b..e2fc94bfd 100644 --- a/docs/content/docs/development/multi-agent/subagents.md +++ b/docs/content/docs/development/multi-agent/subagents.md @@ -433,8 +433,8 @@ parent's. The work happens in your own code or in a remote agent service. The pa only submits to it and waits for the outcome. The framework cannot perform that work for you, because it does not know your protocol. So you write the part that performs one invocation, and the framework supplies everything around it. It runs your code -under durable execution, and after a failover it replays the recorded result instead -of calling your code again. +under durable execution; [Durability and failover](#durability-and-failover) describes +how saved outcomes are reused. What you write is a setup class. Its only job is to carry out one invocation, so it declares no actions and no events of its own. It extends a framework base class and @@ -706,9 +706,9 @@ An async sub-agent talks to a service that runs the task for you. The service ac the work and answers at once, and you query it later for the run's status and result. -An async setup starts the remote run when the caller submits, through one durable -POST. Waiting for the result then polls the run's status until it reaches a terminal -state, and fetches it. +An async setup submits the remote run through durable execution when the caller +submits. Waiting for the result then polls the run's status until it reaches a +terminal state, and fetches it. The base class owns that submit-poll-fetch composition, along with the durable execution and the crash-window recovery. You supply the three transport calls it is @@ -880,10 +880,10 @@ agents: {{< /tabs >}} -The base class also recovers the crash window between "POST sent" and "POST durably -recorded". On recovery it probes the run's status first, and it re-posts only when the -probe reports that the run never started. Override `reconcileSubmitRequest` / -`reconcile_submit_request` only if your service needs a different repair. +When recovery revisits a recorded in-flight submission, the default reconciler +probes the run's status first and re-submits only when the probe reports that the +run never started. Override `reconcileSubmitRequest` / `reconcile_submit_request` +if your service needs a different repair. ## Invoke a sub-agent @@ -1202,26 +1202,30 @@ triggers recovery. ## Durability and failover -A sub-agent call always runs through +Sub-agent calls use [durable execution]({{< ref "docs/development/create-agent/actions#durable-execution" >}}). -Its result is persisted and replayed on recovery, so the work is not repeated after -a failover. There is no durability toggle and no sub-agent timeout option. +To persist invocation outcomes for recovery, configure checkpointing and an +[Action State Store]({{< ref "docs/operations/consistency#configure-an-action-state-store" >}}). +Recovery can then reuse saved outcomes. See +[Consistency]({{< ref "docs/operations/consistency#recovery-guarantees-and-user-responsibilities" >}}) +for replay behavior and limits on external side effects. There is no per-sub-agent +durability toggle or sub-agent timeout option. - Identities are replay-stable. Without explicit ids, the framework assigns `session_id` and `call_id` from the executing action and the sub-agent's resource name, so a replay of the same call gets the same ids. The ids are opaque, so do not parse them. Two sub-agents called from one action never collide, because the resource name is part of the namespace. -- A deferred call runs under the durable id you build in `prepare`, so a replay - re-issues the same logical invocation. A remote keyed on the same values then reuses - the original work instead of starting a second one. The reconciler is optional. - Without it, a crash between the call landing and its result being persisted re-runs - the sub-agent and duplicates its side effects, so supply one whenever the work - cannot be repeated. -- An async call reuses one identity for both its POST and its poll, so your service - sees at most one run per call. -- An internal sub-agent's child actions have their own persisted state, so a replay - reuses the recorded child output instead of running the child again. +- A deferred call uses the durable id you build in `prepare`. Supply a reconciler + when the remote operation cannot be safely repeated; see + [Deferred execution](#deferred-execution). +- An async call uses the same identity for submission, status queries, and result + fetching. Your service must associate them with one logical run and handle + repeated submissions idempotently. The default submission reconciler probes for + that run before re-submitting; see + [Async (pub/sub) execution](#async-pubsub-execution). +- For an internal sub-agent, recovery can reuse saved child Action completions and + outputs. Unfinished child Actions may run again. ## Metrics and Trace Log diff --git a/docs/content/docs/get-started/overview.md b/docs/content/docs/get-started/overview.md index 9d1934640..24cb56785 100644 --- a/docs/content/docs/get-started/overview.md +++ b/docs/content/docs/get-started/overview.md @@ -33,7 +33,7 @@ Like any Agent OS, it manages the core building blocks of an agent - orchestrati The key features of Apache Flink Agents include: - **Massive Scale and Millisecond Latency**: Processes massive-scale event streams in real time, leveraging Flink's distributed processing engine. - **Seamless Data and AI Integration**: Agents interact directly with Flink's DataStream and Table APIs for input and output, enabling a smooth integration of structured data processing and semantic AI capabilities within Flink. -- **Reliable and Exactly-Once Execution**: Ensures exactly-once consistency for agent actions and their side effects by integrating Flink's checkpointing with an external write-ahead log, and supports durable execution - optionally with reconcilers to reconcile in-flight side effects upon failure recovery - so agents run reliably without human supervision. +- **Reliable Execution**: Builds on Flink checkpointing and supports an external Action State Store to reuse saved Action completions and durable call outcomes during recovery. See [Consistency]({{< ref "docs/operations/consistency" >}}) for setup, guarantees, and limits on external side effects. - **Familiar Agent Abstractions**: Leverages well-known AI agent concepts - skills, short/long-term memory, prompts, tools, and dynamic orchestration - making it easy for developers experienced with agent-based systems to quickly adopt and build on Apache Flink Agents without a steep learning curve. - **Multi-Language Support**: Provides Java, Python, and YAML APIs for defining Agents. Actions and supported Resources can be implemented in Java or Python and used across languages. See [Choose APIs and Languages]({{< ref "docs/development/create-agent/choose-apis-and-languages" >}}). - **Rich Ecosystem**: Natively integrates mainstream LLMs, vector stores from diverse providers, and tools or prompts hosted on MCP servers into your agents, while enabling customizable extensions. diff --git a/docs/content/docs/operations/configuration.md b/docs/content/docs/operations/configuration.md index d30592db4..25419f588 100644 --- a/docs/content/docs/operations/configuration.md +++ b/docs/content/docs/operations/configuration.md @@ -1,6 +1,6 @@ --- title: Configuration -weight: 2 +weight: 3 type: docs --- + +When a Flink job recovers from a checkpoint, it restores its state and replays the +input received since that checkpoint. For an Agent, this can mean calling a model, +invoking a tool, or updating an external system again. + +An external Action State Store lets Flink Agents reuse saved Action completions +and durable call outcomes during recovery. Operations whose completion has not +been saved may still run again. The consistency mode you choose determines what +state must remain available for recovery and which work can be reused. + +## Choose a consistency mode + +Flink checkpointing, together with the required source and sink guarantees, lets +each input affect the job's state and output once, even if recovery processes it +again. Flink Agents refers to this checkpoint-based guarantee as **exactly-once +output consistency**. Actions and external calls may still run again, so this +mode requires them to tolerate repetition. + +When recovery also needs to reuse saved work, supplement Flink checkpointing with +an Action State Store. It lets Flink Agents replay saved Action completions +without running their bodies again, and reuse saved durable call outcomes inside +unfinished Actions. Flink Agents calls this **exactly-once action consistency**. +Unfinished Actions may still run again. + +| Configuration | Behavior after a failure | Required setup and cost | +| --- | --- | --- | +| No checkpointing | The job restarts using the source's configured starting positions, without restoring Agent state or progress from a checkpoint. | No checkpoint setup. Suitable when the application can accept lost progress or repeated work. | +| Flink checkpointing | **Exactly-once output consistency.** Flink restores checkpointed state and replays subsequent input. Actions and external calls may run again, and models or queries may return different results. | Enable exactly-once checkpointing and durable checkpoint storage, with the required source and sink guarantees. External calls must tolerate repetition. | +| Flink checkpointing + Action State Store | **Exactly-once action consistency.** Flink Agents also reuses saved Action completions. Unfinished Actions may run again, reusing any saved durable call outcomes as they do so. | Configure checkpointing and a Kafka or Fluss Action State Store. Saving Action completions and durable call outcomes requires synchronous backend writes. Retain both the checkpoints and the required backend data. Use idempotency or reconciliation to protect external effects. | + +## Required setup + +Enable exactly-once checkpointing in the hosting Flink or PyFlink job, with source +and sink connectors that provide the required +[fault tolerance guarantees](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/connectors/datastream/guarantees/). +Follow Flink's [checkpointing guide](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/dev/datastream/fault-tolerance/checkpointing/) +for configuration and checkpoint storage, and +[checkpoint recovery guide](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/ops/state/checkpoints/#resuming-from-a-retained-checkpoint) +for restore procedures. Use the Flink documentation for the version you deploy. + +### Configure an Action State Store + +For exactly-once action consistency, also configure Kafka or Fluss as the Action +State Store backend. Set the backend options before building the Agent pipeline. + +The Python and Java examples use `agents_env` and `agentsEnv`, respectively, +created as described in [Integrate with Flink]({{< ref "docs/development/integrate_with_flink" >}}). +The YAML examples show the `agent:` section of the hosting job's Flink +configuration. + +Each logical Agent operator needs its **own Kafka topic or Fluss table**, shared +only by that operator's subtasks. + +#### Kafka + +Set the backend to `kafka` and provide the broker address and a dedicated topic: + +{{< tabs "Kafka Action State Store" >}} +{{< tab "Python" >}} +```python +from flink_agents.api.core_options import AgentConfigOptions + +config = agents_env.get_config() +config.set(AgentConfigOptions.ACTION_STATE_STORE_BACKEND, "kafka") +config.set(AgentConfigOptions.KAFKA_BOOTSTRAP_SERVERS, "kafka:9092") +config.set(AgentConfigOptions.KAFKA_ACTION_STATE_TOPIC, "orders-agent-action-state") +``` +{{< /tab >}} +{{< tab "Java" >}} +```java +import org.apache.flink.agents.api.configuration.AgentConfigOptions; +import org.apache.flink.agents.api.configuration.Configuration; + +Configuration config = agentsEnv.getConfig(); +config.set(AgentConfigOptions.ACTION_STATE_STORE_BACKEND, "kafka"); +config.set(AgentConfigOptions.KAFKA_BOOTSTRAP_SERVERS, "kafka:9092"); +config.set(AgentConfigOptions.KAFKA_ACTION_STATE_TOPIC, "orders-agent-action-state"); +``` +{{< /tab >}} +{{< tab "YAML" >}} +```yaml +agent: + actionStateStoreBackend: kafka + kafkaBootstrapServers: "kafka:9092" + kafkaActionStateTopic: orders-agent-action-state +``` +{{< /tab >}} +{{< /tabs >}} + +The store creates the topic if it does not exist, using 64 partitions and a +replication factor of 1 by default. If the topic already exists, the store reuses +it without changing its settings. For retention requirements and cleanup options, +see [Kafka retention and cleanup](#kafka-retention-and-cleanup). + +#### Fluss + +Set the backend to `fluss` and provide the server address and a dedicated table: + +{{< tabs "Fluss Action State Store" >}} +{{< tab "Python" >}} +```python +from flink_agents.api.core_options import AgentConfigOptions + +config = agents_env.get_config() +config.set(AgentConfigOptions.ACTION_STATE_STORE_BACKEND, "fluss") +config.set(AgentConfigOptions.FLUSS_BOOTSTRAP_SERVERS, "fluss:9123") +config.set(AgentConfigOptions.FLUSS_ACTION_STATE_DATABASE, "flink_agents") +config.set(AgentConfigOptions.FLUSS_ACTION_STATE_TABLE, "orders_agent_action_state") +``` +{{< /tab >}} +{{< tab "Java" >}} +```java +import org.apache.flink.agents.api.configuration.AgentConfigOptions; +import org.apache.flink.agents.api.configuration.Configuration; + +Configuration config = agentsEnv.getConfig(); +config.set(AgentConfigOptions.ACTION_STATE_STORE_BACKEND, "fluss"); +config.set(AgentConfigOptions.FLUSS_BOOTSTRAP_SERVERS, "fluss:9123"); +config.set(AgentConfigOptions.FLUSS_ACTION_STATE_DATABASE, "flink_agents"); +config.set(AgentConfigOptions.FLUSS_ACTION_STATE_TABLE, "orders_agent_action_state"); +``` +{{< /tab >}} +{{< tab "YAML" >}} +```yaml +agent: + actionStateStoreBackend: fluss + flussBootstrapServers: "fluss:9123" + flussActionStateDatabase: flink_agents + flussActionStateTable: orders_agent_action_state +``` +{{< /tab >}} +{{< /tabs >}} + +The store creates the database and append-only log table if they do not exist. +It uses the `flink_agents` database and 64 buckets for a new table by default. +Configure the table's log retention to keep the data needed by your recovery +points; the store does not set retention when it creates the table. For an +authenticated Fluss cluster, also set the appropriate authentication options. + +See [Configuration]({{< ref "docs/operations/configuration" >}}) for option defaults, +authentication settings, and configuration loading and precedence. + +## Recovery guarantees and user responsibilities + +### Recovery flow + +Exactly-once action consistency reuses work only after the relevant execution +records have been saved. When a job restores a checkpoint or savepoint using the +corresponding Action State Store, it handles each Action as follows: + +1. **Check for a saved Action completion.** If one is available, replay its saved + Sensory/Short-Term Memory updates and output Events. The Events keep their + original Event IDs. The Action body is skipped, completing recovery for that + Action. +2. **Rerun the body of an unfinished Action.** Code and ordinary calls within it + may repeat. When execution reaches a durable call, check for a matching call + record: + + - **Saved success or failure:** replay the saved outcome without calling the + function again. + - **Recorded in-flight call:** run its reconciler if one is configured. + Otherwise, execute the function again. + - **No matching record:** execute the function normally. + +### Conditions for reusing saved records + +Saved records must match the same inputs and calls during replay: + +- **Preserve input order for each key.** Recovery must receive each key's inputs + in the same order as the original execution. +- **For custom Actions, keep the durable call sequence stable.** Use durable + execution for calls whose saved outcomes should be reused. Rerunning an + unfinished Action must reach those calls in the same order, with the same + inputs. See [Durable Execution]({{< ref "docs/development/create-agent/actions#durable-execution" >}}) + for call identity, serialization, and Action-context requirements. + +Before restoring after changes to keys, Action names, result types, application +code, or versions, review +[Upgrades and migration](#upgrades-and-migration). + +### Limits on external side effects + +An external operation can finish before its outcome is saved. If the job fails +in that interval, recovery may execute the operation again. The Action State +Store cannot commit the external operation and its saved outcome as a single +transaction. + +For operations that change external state, use the service's idempotency support +or, for a durable call, provide a reconciler that can recover the earlier outcome. +See [Reconciling an in-flight call]({{< ref "docs/development/create-agent/actions#reconciling-an-in-flight-call" >}}) +for the API and requirements. + +## Action State Store retention and cleanup + +For every checkpoint or savepoint you intend to restore, retain both the snapshot +and its corresponding Action State Store records. Before changing retention or +cleanup settings, decide which recovery points must remain available and select +the oldest one you intend to keep. + +{{< hint warning >}} +Hashing business keys does not encrypt Action state. Saved Events, Memory updates, +and durable call outcomes may contain sensitive data. Protect access to these +records. +{{< /hint >}} + +### Kafka retention and cleanup + +Topics created by the store use `cleanup.policy=compact,delete`, +`retention.ms=-1`, and `retention.bytes=-1`. The `-1` values disable the time and +size limits for the `delete` policy. Compaction can still reclaim older values +for the same key. + +If you reuse an existing topic, check that its retention settings keep the +records needed by every checkpoint or savepoint you intend to restore. + +A completed checkpoint already includes the Agent's progress for the work it +covers. The corresponding execution records are no longer needed to restore that +checkpoint, but earlier recovery points may still need them. Choose a cleanup +method based on the recovery history you need to retain: + +| Method | Cleanup behavior | When to use it | Recovery impact | +| --- | --- | --- | --- | +| No tombstones (default) | With `kafkaActionStateTombstoneEnabled=false`, completed work is removed from the in-memory cache, while its Kafka records remain. | Kafka storage growth is acceptable and you need to retain older recovery points. | Pruning alone preserves older recovery points. Kafka retention or manual deletion can still remove required data. | +| Per-key tombstones | The store automatically writes deletion markers for execution records covered by each completed checkpoint. Kafka can reclaim those records through background compaction. | The job only needs the latest completed checkpoint, or repeating affected Actions is acceptable. | The triggering checkpoint remains usable. Restoring an earlier checkpoint or savepoint may repeat completed Actions. | +| Checkpoint-aligned cleanup | An administrator selects the oldest recovery point to retain. The tool records a cleanup boundary in a separate control topic, then deletes older Kafka data that point no longer needs. | You need to retain a selected checkpoint or savepoint while reclaiming older recovery history. | The selected point remains usable. With the control topic configured, recovery points requiring data below the boundary are rejected. | + +Per-key tombstones and checkpoint-aligned cleanup cannot be enabled together. +A committed cleanup boundary takes effect even if physical deletion has not +finished. + +#### Cleanup steps + +For per-key tombstones, set `kafkaActionStateTombstoneEnabled=true` before starting +the job. + +For checkpoint-aligned cleanup, follow the procedure below. +`KafkaActionStateCleanupTool` provides the same cleanup procedure for Java and +Python jobs. Use the full Flink Agents distribution JAR for your Flink version +(`flink-agents-dist-flink--.jar`, without the `-thin` +classifier), the matching Flink State Processor API JAR, and Flink's `lib/` on the +classpath. Run the tool where the recovery point and Kafka are accessible. + +If the job has used per-key tombstones, first follow +[Switching from per-key tombstones](#switching-from-per-key-tombstones). + +For the first cleanup of an existing job: + +1. Select the oldest completed checkpoint or savepoint you intend to retain, and + stop the job while keeping that recovery point available. Identify the Agent + operator by its UID or UID hash in that snapshot. The selected point must + contain versioned Kafka recovery markers. + See the [configuration reference]({{< ref "docs/operations/configuration#checkpoint-aligned-kafka-cleanup" >}}) + for recovery marker requirements. +2. Generate a cleanup plan. Replace the JAR and recovery-point placeholders below; + use `--operator-uid` instead of `--operator-uid-hash` when specifying a UID: + + ```bash + java -cp "$FLINK_HOME/lib/*::" \ + org.apache.flink.agents.runtime.actionstate.KafkaActionStateCleanupTool \ + plan --checkpoint "" \ + --operator-uid-hash "" \ + --output cleanup-plan.json + ``` + +3. Check that the plan's `sourceRecoveryPoint` and `topic` match the selected + snapshot and this Agent operator's Action State Store. Keep the plan file, and + ensure the data topic uses the retention settings listed above. +4. Apply the plan using a separate control topic: + + ```bash + java -cp "$FLINK_HOME/lib/*::" \ + org.apache.flink.agents.runtime.actionstate.KafkaActionStateCleanupTool \ + apply --plan cleanup-plan.json --bootstrap-servers "kafka:9092" \ + --control-topic "orders-agent-action-state-cleanup" + ``` + + Dedicate the control topic to this Agent operator's data topic. The tool can + create it; see the [configuration reference]({{< ref "docs/operations/configuration#checkpoint-aligned-kafka-cleanup" >}}) + for existing-topic requirements and optional parameters. +5. Wait for the command to succeed and print `Cleanup plan is APPLIED`. + Restart from the **same selected recovery point**, with + `kafkaActionStateCleanupControlTopic` set to that control topic and + `kafkaActionStateTombstoneEnabled=false`. + +If this first cleanup fails or is interrupted, keep the job stopped and retry +`apply` with the same plan. The boundary may already have been committed, making +earlier recovery points unavailable. + +Set the control-topic option after the first successful `apply`. Startup or +recovery fails if the configured topic is missing or has no committed boundary. + +With boundary enforcement enabled, you can apply later cleanup plans while the job +runs, provided they only advance the boundary. Apply one plan at a time, and keep +the data topic and its partition count unchanged during cleanup. + +{{< hint warning >}} +Keep the same control-topic setting and its records for every subsequent run and +restore of this recovery history. Omitting or changing that setting removes the +protection against restoring an older point. With the control topic configured, +the job also rejects a fresh start without a recovery marker. +{{< /hint >}} + +#### Switching from per-key tombstones + +For a job that previously emitted tombstones: + +1. Stop the job cleanly and wait for pending tombstone writes to finish. +2. Restore the newest recovery point that is still valid under the tombstone + policy. Set `kafkaActionStateTombstoneEnabled=false` and leave the cleanup + control-topic option unset. +3. Complete and retain a new checkpoint or savepoint, then stop the job again. +4. Use that new recovery point for the first cleanup plan and apply operation. +5. Restart from the same point with the cleanup control topic configured. + +Use this new recovery point because an older one may still require records +affected by past tombstones. + +### Fluss retention + +Fluss log retention controls the physical cleanup of Action state. The per-key +tombstones and checkpoint-aligned cleanup described above apply only to the Kafka +Action State Store. + +## Upgrades and migration + +### Check recovery conditions before a change + +Before restoring state after a change, check both Flink's state compatibility +requirements and whether the application can still use its saved Action results. +Flink may be able to read a checkpoint even when those results are no longer +compatible with the application. + +| Planned change | What to check | +| --- | --- | +| Flink Agents or Flink version | Follow the target release's migration guidance and Flink's upgrade requirements. A job starting successfully does not confirm that recovery across those versions is supported. | +| Key type or serializer | Keep the original key type and serializer configuration when reusing Action state. Custom key serializers must produce deterministic bytes. Changes to the serialized key bytes are not always detected. | +| Action name | A renamed Action no longer matches its old completion. Keeping the same name allows recovery to reuse old output without running the new body. Check whether that saved output is still valid for the new code. | +| User code, Events, Memory, or durable results | Check that saved outputs and values can still be read and used, and that unfinished Actions reach the same logical calls. Runtime format checks do not verify compatibility with the application's logic. | +| Action State Store backend | When reusing checkpoints or savepoints, preserve the original [Kafka topic]({{< ref "docs/operations/configuration" >}}#changes-to-an-existing-kafka-store) or [Fluss table]({{< ref "docs/operations/configuration" >}}#changes-to-an-existing-fluss-store), its partition or bucket layout, and [the records needed for recovery](#action-state-store-retention-and-cleanup). | + +Use Flink's +[application and version upgrade guidance](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/ops/upgrading/) +for the hosting job. The checks above identify recovery requirements and known +risks. They do not establish which Flink Agents and Flink version combinations are +supported. + +### Starting without existing state + +If the target release requires starting without restoring an existing checkpoint +or savepoint, configure the source's starting position before starting the new job. +If the job will reprocess input handled by the previous version, ensure external +operations are idempotent or deduplicated to avoid repeating their effects. + +### Migration notes + +#### Flink Agents 0.4 + +Flink Agents 0.4 changes the Action State Store record key format and does not +support records written by earlier versions. When upgrading from an earlier +version, use a new Kafka topic or Fluss table and +[start without restoring an earlier checkpoint or savepoint](#starting-without-existing-state). diff --git a/docs/content/docs/operations/deployment.md b/docs/content/docs/operations/deployment.md index 8d9ee4e96..de81ba1c0 100644 --- a/docs/content/docs/operations/deployment.md +++ b/docs/content/docs/operations/deployment.md @@ -84,60 +84,11 @@ For all available options, see the [Flink CLI documentation](https://nightlies.a Now you should see a Flink job submitted to the Flink Cluster in Flink web UI (typically accessible at http://<jobmanagerHost>:8081). -## Consistency Guarantees - -Flink Agents leverage Flink’s checkpointing to provide **exactly-once output consistency**. Exactly-once output consistency does not mean each event is processed only once; it means that every event will affect the state and output result exactly once. This is sufficient for typical data-processing use cases where the processing of data does not cause any external side effects. - -However, actions in Flink Agents can trigger external side effects (for example, sending an email or invoking an external API). These side effects should not be executed more than once. To address this, Flink Agents support **exactly-once action consistency** by leveraging an external action state store. - -### Exactly-Once Output Consistency - -When a job runs on a Flink cluster, Flink periodically takes checkpoints. See [Flink Checkpointing](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/ops/state/checkpoints/) for details. -After recovery from a checkpoint, Flink Agents reprocess events that arrived after that checkpoint. As a result, any actions triggered by those events may be executed again. - -### Exactly-Once Action Consistency - -To enable exactly-once action consistency, configure an external action state store. -Flink Agents record state for each Action in this store. After recovering from a -checkpoint, the framework reuses Actions whose completion was persisted, including -their stored Sensory/Short-Term Memory updates and output Events. Unfinished Actions -may execute again. - -When a completed Action is reused during recovery, its stored output Events keep their original Event IDs. The runtime records their relationships to the current trigger Event in Trace Log `entityMetadata`. - -The same store also saves individual call outcomes for -[Durable Execution]({{< ref "docs/development/create-agent/actions#durable-execution" >}}). -When an unfinished Action runs again, durable calls can reuse persisted outcomes. -An external operation can finish before its outcome is saved, so recovery may repeat -that operation. Use idempotent operations or reconciliation for external side effects; -see [Recovery Behavior]({{< ref "docs/development/create-agent/actions#recovery-behavior" >}}). - -{{< hint info >}} -**Note**: Currently, Kafka and Fluss are supported as the external action state store. -{{< /hint >}} - -{{< hint warning >}} -The action-state key format changed in Flink Agents 0.4 and action-state records written by earlier versions are unsupported. When upgrading, use a fresh Kafka topic or Fluss table and start without an older checkpoint or savepoint. Recovery validates the current fields, but cannot reliably distinguish every older record from the current format. + + -Action-state keys contain a digest of the serialized key. Changing key types or serializer configuration when recovering existing state is unsupported, even if Flink accepts the change for keyed state: the same key can serialize to different bytes, causing recovery to miss completed actions and repeat their side effects. These changes are not detected by the action-state store. Preserve the original key types and serializer configuration for recovery. Custom key serializers must produce deterministic bytes. - -Action-state keys identify an action by its name. Keeping an action's name across a code change lets recovery reuse its completed results; renaming an action makes its recorded results unreachable, so it is executed again on the next recovery. - -Dedicate each Kafka topic or Fluss table to one logical Flink Agents operator, shared by that operator's subtasks. Action-state keys do not contain a job or operator namespace, so sharing a backend between logical operators can allow otherwise identical records to collide. -{{< /hint >}} - -The business-key component stored in the backend is a SHA-256 digest rather than the serialized key itself. This keeps record keys bounded and avoids embedding raw key bytes, but it is not encryption; protect the action-state backend with appropriate access controls. - -See [Action State Store Configuration]({{< ref "docs/operations/configuration#action-state-store" >}}) for configuration options. - -{{< hint warning >}} -**Note**: Enabling Kafka action-state tombstones can invalidate checkpoints or savepoints older than the prune and cause completed actions to execute again. See [Action State Store Configuration]({{< ref "docs/operations/configuration#action-state-store" >}}) for the recovery trade-off. -{{< /hint >}} - -{{< hint warning >}} -**Note**: A committed checkpoint-aligned Kafka cleanup boundary permanently makes older recovery points unsupported. Every subsequent run and restore must use the same cleanup control topic so the runtime can reject those recovery points before replay. See [Checkpoint-aligned Kafka cleanup]({{< ref "docs/operations/configuration#checkpoint-aligned-kafka-cleanup" >}}). -{{< /hint >}} +## Consistency Guarantees -{{< hint info >}} -**Note**: Exactly-once action consistency is guaranteed only if, after recovering from the same checkpoint, inputs for each key arrive in the same order as before recovery. If this ordering requirement is not met, the system falls back to exactly-once output consistency. -{{< /hint >}} +Before submitting the job, choose and configure its consistency mode. See +[Consistency]({{< ref "docs/operations/consistency" >}}) for required setup, +recovery guarantees, retention and cleanup, and upgrade guidance. diff --git a/docs/content/docs/operations/monitoring.md b/docs/content/docs/operations/monitoring.md index 3e67f447b..93b762bc7 100644 --- a/docs/content/docs/operations/monitoring.md +++ b/docs/content/docs/operations/monitoring.md @@ -1,6 +1,6 @@ --- title: Monitoring -weight: 3 +weight: 4 type: docs --- When a Flink job recovers from a checkpoint, it restores its state and replays the -input received since that checkpoint. For an Agent, this can mean calling a model, -invoking a tool, or updating an external system again. - -The consistency mode you choose determines what state must remain available for -recovery and which work can be reused. +input received since that checkpoint. An Agent may therefore call a model, invoke +a tool, or update an external system again. The consistency mode you choose +determines what state must remain available for recovery and which work can be +reused. ## Choose a consistency mode @@ -35,9 +34,9 @@ With exactly-once checkpointing and the required source and sink guarantees, each input affects the job's state and output once, even if recovery processes it again. Flink Agents calls this **exactly-once output consistency**. -Adding an external Action State Store lets recovery also reuse saved Action -completions and durable call outcomes. Flink Agents calls this **exactly-once -action consistency**. +Checkpoint recovery may still rerun Actions and external calls, so you can add an +external Action State Store to reuse saved Action completions and durable call +outcomes. Flink Agents calls this **exactly-once action consistency**. | Configuration | Behavior after a failure | Requirements and trade-offs | | --- | --- | --- | @@ -57,14 +56,14 @@ for restore procedures. Use the Flink documentation for the version you deploy. ### Configure an Action State Store -Configure the Kafka or Fluss backend before building the Agent pipeline. - -Create `agents_env` (Python) or `agentsEnv` (Java) as described in -[Integrate with Flink]({{< ref "docs/development/integrate_with_flink" >}}). -The YAML examples show the `agent:` section of the job's Flink configuration. +Before building the Agent pipeline, configure Kafka or Fluss as the Action State +Store backend. Each logical Agent operator needs its **own Kafka topic or Fluss +table**, shared only by that operator's subtasks. -Each logical Agent operator needs its **own Kafka topic or Fluss table**, shared -only by that operator's subtasks. +The Python and Java examples below use `agents_env` and `agentsEnv`, respectively, +created as described in [Integrate with Flink]({{< ref "docs/development/integrate_with_flink" >}}). +The YAML examples show the same settings in the `agent:` section of the job's +Flink configuration. #### Kafka @@ -102,8 +101,9 @@ agent: {{< /tab >}} {{< /tabs >}} -The store creates the topic if needed, with 64 partitions and a replication -factor of 1 by default. It reuses existing topics without changing their settings. +If the topic does not exist, the store creates it with 64 partitions and a +replication factor of 1 by default. Otherwise, it reuses the existing topic without +changing its settings. For retention requirements and cleanup options, see [Kafka retention and cleanup](#kafka-retention-and-cleanup). @@ -147,13 +147,13 @@ agent: {{< /tabs >}} The store creates the database and append-only log table if needed. By default, -it uses the `flink_agents` database and creates new tables with 64 buckets. It does -not set log retention when creating the table; configure retention to preserve -the data needed by your recovery points. If the cluster requires authentication, -also configure the Fluss authentication options. +it uses the `flink_agents` database and creates new tables with 64 buckets. -See [Configuration]({{< ref "docs/operations/configuration" >}}) for option defaults, -authentication settings, and configuration loading and precedence. +The store does not set log retention when creating a table, so configure retention +to preserve the records required by your recovery points. If the cluster requires +authentication, configure the Fluss authentication options as well. See +[Configuration]({{< ref "docs/operations/configuration" >}}) for these settings, +option defaults, and configuration loading and precedence. ## Recovery guarantees and user responsibilities @@ -162,11 +162,13 @@ authentication settings, and configuration loading and precedence. When restoring a checkpoint or savepoint with its corresponding Action State Store, Flink Agents handles each Action as follows: -1. **Check for a saved Action completion.** If one is available, replay its saved - Sensory/Short-Term Memory updates and output Events. The Events keep their - original Event IDs. Skip the Action body. -2. **Rerun an unfinished Action.** Ordinary calls may repeat. At each durable call, - check for a matching record: +1. **Check for a saved Action completion.** If one is available, Flink Agents + replays its saved Sensory/Short-Term Memory updates and output Events without + running the Action body again. The replayed Events keep their original Event + IDs. +2. **Rerun an unfinished Action.** If no completion is found, Flink Agents runs the + Action body again. Ordinary calls may therefore repeat, but the framework + checks each durable call for a matching record before executing it: - **Saved success or failure:** replay the saved outcome without calling the function again. @@ -190,10 +192,9 @@ application code, or versions, review [Upgrades and migration](#upgrades-and-mig ### Limits on external side effects -An external operation can finish before its outcome is saved. If the job fails -in that interval, recovery may execute the operation again. The Action State -Store cannot commit the external operation and its saved outcome as a single -transaction. +The Action State Store cannot commit an external operation and its saved outcome +as a single transaction. If the operation finishes but the job fails before its +outcome is saved, recovery may execute it again. For operations that change external state, use the service's idempotency support or, for a durable call, provide a reconciler that can recover the original outcome. @@ -222,10 +223,10 @@ for the same key. For an existing topic, check that its retention settings preserve the records needed by every recovery point you intend to keep. -A completed checkpoint includes the Agent's progress for the work it covers. -Those execution records are no longer needed to restore that checkpoint, but -earlier recovery points may still need them. Choose a cleanup method based on -the recovery history you need to retain: +A completed checkpoint includes the Agent's progress for the work it covers, +which means those execution records are no longer needed to restore it. Because +earlier recovery points may still need the same records, choose a cleanup method +based on the recovery history you need to retain: | Method | Cleanup behavior | When to use it | Recovery impact | | --- | --- | --- | --- | @@ -291,19 +292,20 @@ For the first cleanup of an existing job: **same selected recovery point**. Startup or recovery fails if the control topic is missing or has no committed boundary. -If this first cleanup fails or is interrupted, keep the job stopped and retry -`apply` with the same plan. The boundary may already have been committed, making -earlier recovery points unavailable. +If the first cleanup fails or is interrupted, the boundary may already have been +committed, making earlier recovery points unavailable. Keep the job stopped and +retry `apply` with the same plan. -With boundary enforcement enabled, you can apply later cleanup plans while the job -runs, provided they only advance the boundary. Apply one plan at a time, and keep -the data topic and its partition count unchanged during cleanup. +Once boundary enforcement is enabled, you can apply later cleanup plans while the +job runs, provided they only advance the boundary. Apply one plan at a time, +keeping the data topic and its partition count unchanged during cleanup. {{< hint warning >}} -Keep the same control-topic setting and retain its records whenever you run or -restore the job using this recovery history. Omitting or changing the setting -removes the protection against restoring an older point. With the control topic -configured, the job also rejects a fresh start without a recovery marker. +Keep the job configured to use the same cleanup control topic, and retain that +topic's records for every subsequent run and restore of this recovery history. +The job needs this configuration to enforce the cleanup boundary; omitting or +changing it removes that protection. With the control topic configured, the job +also rejects a fresh start without a recovery marker. {{< /hint >}} #### Switching from per-key tombstones @@ -353,8 +355,9 @@ supported. If the target release requires a fresh start without restoring a checkpoint or savepoint, configure the source's starting position before launching the new job. -If it will reprocess input handled by the previous version, use idempotency or -deduplication to prevent repeated external effects. +Depending on that starting position, the new job may reprocess input handled by +the previous version. If it does, use idempotency or deduplication to prevent +repeated external effects. ### Migration notes From 85103d3f0724f1ad3937bcba8bb765ce2d085ccc Mon Sep 17 00:00:00 2001 From: joeyutong Date: Tue, 20 Oct 2026 11:48:39 +0800 Subject: [PATCH 08/13] [docs] Point Fluss cleanup to native retention settings Explain how to manage Fluss records after identifying the absence of Kafka-style cleanup mechanisms in Flink Agents. Generated-by: Codex CLI 0.163.0-alpha.10.2 (GPT-6) Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 3/3 AI-Contributed/UT: 0/0 --- docs/content/docs/operations/consistency.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/docs/content/docs/operations/consistency.md b/docs/content/docs/operations/consistency.md index 493d4ade1..cd37ce322 100644 --- a/docs/content/docs/operations/consistency.md +++ b/docs/content/docs/operations/consistency.md @@ -326,7 +326,8 @@ affected by past tombstones. ### Fluss retention Flink Agents currently provides no equivalent of Kafka's per-key tombstones or -checkpoint-aligned cleanup for the Fluss Action State Store. +checkpoint-aligned cleanup for the Fluss Action State Store. Instead, use Fluss's +own log retention settings to manage record retention and cleanup. ## Upgrades and migration From 81f7c9002c6f23ae0e16ef1536d1f95046fe79c8 Mon Sep 17 00:00:00 2001 From: joeyutong Date: Tue, 20 Oct 2026 12:45:30 +0800 Subject: [PATCH 09/13] [docs] Simplify the 0.4 migration instructions Keep the fresh-start requirement in the 0.4 migration note and remove the redundant subsection and source-position guidance. Generated-by: Codex CLI 0.163.0-alpha.10.2 (GPT-6) Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 12/12 AI-Contributed/UT: 0/0 --- docs/content/docs/operations/consistency.md | 12 ++---------- 1 file changed, 2 insertions(+), 10 deletions(-) diff --git a/docs/content/docs/operations/consistency.md b/docs/content/docs/operations/consistency.md index cd37ce322..ce584c017 100644 --- a/docs/content/docs/operations/consistency.md +++ b/docs/content/docs/operations/consistency.md @@ -352,19 +352,11 @@ for the hosting job. The checks above identify recovery requirements and known risks. They do not establish which Flink Agents and Flink version combinations are supported. -### Starting without existing state - -If the target release requires a fresh start without restoring a checkpoint or -savepoint, configure the source's starting position before launching the new job. -Depending on that starting position, the new job may reprocess input handled by -the previous version. If it does, use idempotency or deduplication to prevent -repeated external effects. - ### Migration notes #### Flink Agents 0.4 Flink Agents 0.4 changes the Action State Store record key format and does not support records written by earlier versions. When upgrading from an earlier -version, use a new Kafka topic or Fluss table and -[start without restoring an earlier checkpoint or savepoint](#starting-without-existing-state). +version, use a new Kafka topic or Fluss table and submit the new job without +restoring a checkpoint or savepoint from the previous version. From dc9483f06a0505dc24c444d38bd7b7c88d88cd33 Mon Sep 17 00:00:00 2001 From: joeyutong Date: Tue, 20 Oct 2026 14:37:03 +0800 Subject: [PATCH 10/13] [docs] Condense checkpoint-aligned cleanup guidance Keep the cleanup procedure focused on four key steps and recovery restrictions. Place complete commands and classpath requirements in a collapsed examples block. Generated-by: Codex CLI 0.163.0-alpha.10.2 (GPT-6) Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 113/113 AI-Contributed/UT: 0/0 --- docs/content/docs/operations/consistency.md | 113 ++++++++++---------- 1 file changed, 57 insertions(+), 56 deletions(-) diff --git a/docs/content/docs/operations/consistency.md b/docs/content/docs/operations/consistency.md index ce584c017..5f18d7984 100644 --- a/docs/content/docs/operations/consistency.md +++ b/docs/content/docs/operations/consistency.md @@ -244,70 +244,71 @@ For per-key tombstones, set `kafkaActionStateTombstoneEnabled=true` before start the job. For checkpoint-aligned cleanup, use `KafkaActionStateCleanupTool` for both Java -and Python jobs. Use the full Flink Agents distribution JAR for your Flink version -(`flink-agents-dist-flink--.jar`, without the `-thin` -classifier), the matching Flink State Processor API JAR, and Flink's `lib/` on the -classpath. Run the tool where the recovery point and Kafka are accessible. - -If the job has used per-key tombstones, first follow +and Python jobs. Review the [configuration prerequisites]({{< ref "docs/operations/configuration#checkpoint-aligned-kafka-cleanup" >}}) +before running the tool. If the job has used per-key tombstones, first follow [Switching from per-key tombstones](#switching-from-per-key-tombstones). For the first cleanup of an existing job: -1. Select the oldest completed checkpoint or savepoint you intend to retain, and - stop the job while keeping that recovery point available. Identify the Agent - operator by its UID or UID hash in that snapshot. The selected point must - contain versioned Kafka recovery markers. - See the [configuration reference]({{< ref "docs/operations/configuration#checkpoint-aligned-kafka-cleanup" >}}) - for recovery marker requirements. -2. Generate a cleanup plan. Replace the JAR and recovery-point placeholders below; - use `--operator-uid` instead of `--operator-uid-hash` when specifying a UID: - - ```bash - java -cp "$FLINK_HOME/lib/*::" \ - org.apache.flink.agents.runtime.actionstate.KafkaActionStateCleanupTool \ - plan --checkpoint "" \ - --operator-uid-hash "" \ - --output cleanup-plan.json - ``` - -3. Check that the plan's `sourceRecoveryPoint` and `topic` match the selected - snapshot and this Agent operator's Action State Store. Keep the plan file, and - ensure the data topic uses the retention settings listed above. -4. Apply the plan using a separate control topic: - - ```bash - java -cp "$FLINK_HOME/lib/*::" \ - org.apache.flink.agents.runtime.actionstate.KafkaActionStateCleanupTool \ - apply --plan cleanup-plan.json --bootstrap-servers "kafka:9092" \ - --control-topic "orders-agent-action-state-cleanup" - ``` - - Dedicate the control topic to this Agent operator's data topic. The tool can - create it; see the [configuration reference]({{< ref "docs/operations/configuration#checkpoint-aligned-kafka-cleanup" >}}) - for existing-topic requirements and optional parameters. -5. After `apply` succeeds and prints `Cleanup plan is APPLIED`, set - `kafkaActionStateCleanupControlTopic` to the control topic and - `kafkaActionStateTombstoneEnabled=false`, then restart from the - **same selected recovery point**. Startup or recovery fails if the control - topic is missing or has no committed boundary. - -If the first cleanup fails or is interrupted, the boundary may already have been -committed, making earlier recovery points unavailable. Keep the job stopped and -retry `apply` with the same plan. - -Once boundary enforcement is enabled, you can apply later cleanup plans while the -job runs, provided they only advance the boundary. Apply one plan at a time, -keeping the data topic and its partition count unchanged during cleanup. +1. Select the oldest completed checkpoint or savepoint to retain, and stop the + job while keeping that snapshot available. +2. Generate a cleanup plan for the Agent operator in that snapshot, and check + that it matches the selected recovery point and Action State Store. +3. Apply the plan with a separate cleanup control topic dedicated to that Action + State Store. +4. After `apply` succeeds, set `kafkaActionStateCleanupControlTopic` to that topic + and `kafkaActionStateTombstoneEnabled=false`, then restart from the + **same selected recovery point**. + +If the first cleanup fails or is interrupted, keep the job stopped and retry +`apply` with the same plan: the boundary may already have been committed, making +earlier recovery points unavailable. + +Once boundary enforcement is enabled, later plans may be applied while the job +runs. Apply one plan at a time, only advance the boundary, and keep the data topic +and its partition count unchanged during cleanup. {{< hint warning >}} -Keep the job configured to use the same cleanup control topic, and retain that -topic's records for every subsequent run and restore of this recovery history. -The job needs this configuration to enforce the cleanup boundary; omitting or -changing it removes that protection. With the control topic configured, the job -also rejects a fresh start without a recovery marker. +Keep the same cleanup control topic configured and retain its committed records +for every subsequent run and restore of this recovery history. Omitting or +changing the topic removes boundary enforcement; with it configured, the job also +rejects a fresh start without a recovery marker. {{< /hint >}} +{{< details title="Command examples" >}} + +Run the tool where the recovery point and Kafka are accessible. The classpath +must include the full Flink Agents distribution JAR for your Flink version +(`flink-agents-dist-flink--.jar`, without the `-thin` +classifier), the matching Flink State Processor API JAR, and Flink's `lib/`. + +Generate a plan with the Agent operator's UID hash in the selected snapshot. +Replace the placeholders below, or use `--operator-uid` instead of +`--operator-uid-hash` when specifying a UID: + +```bash +java -cp "$FLINK_HOME/lib/*::" \ + org.apache.flink.agents.runtime.actionstate.KafkaActionStateCleanupTool \ + plan --checkpoint "" \ + --operator-uid-hash "" \ + --output cleanup-plan.json +``` + +After checking the plan, apply it with a separate control topic. The tool can +create the topic; existing-topic requirements and optional parameters are listed +in the [configuration reference]({{< ref "docs/operations/configuration#checkpoint-aligned-kafka-cleanup" >}}). + +```bash +java -cp "$FLINK_HOME/lib/*::" \ + org.apache.flink.agents.runtime.actionstate.KafkaActionStateCleanupTool \ + apply --plan cleanup-plan.json --bootstrap-servers "kafka:9092" \ + --control-topic "orders-agent-action-state-cleanup" +``` + +A successful apply prints `Cleanup plan is APPLIED`. + +{{< /details >}} + #### Switching from per-key tombstones To switch a job from per-key tombstones to checkpoint-aligned cleanup: From a58e5d655b633fa12bb93c0eb7ba91b02a2d0e1c Mon Sep 17 00:00:00 2001 From: joeyutong Date: Tue, 20 Oct 2026 14:42:21 +0800 Subject: [PATCH 11/13] [docs] Keep cleanup commands with the main procedure Place the collapsed command examples immediately after the four cleanup steps, ahead of failure handling and recovery restrictions. Generated-by: Codex CLI 0.163.0-alpha.10.2 (GPT-6) Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 30/30 AI-Contributed/UT: 0/0 --- docs/content/docs/operations/consistency.md | 30 ++++++++++----------- 1 file changed, 15 insertions(+), 15 deletions(-) diff --git a/docs/content/docs/operations/consistency.md b/docs/content/docs/operations/consistency.md index 5f18d7984..f77f92d4e 100644 --- a/docs/content/docs/operations/consistency.md +++ b/docs/content/docs/operations/consistency.md @@ -260,21 +260,6 @@ For the first cleanup of an existing job: and `kafkaActionStateTombstoneEnabled=false`, then restart from the **same selected recovery point**. -If the first cleanup fails or is interrupted, keep the job stopped and retry -`apply` with the same plan: the boundary may already have been committed, making -earlier recovery points unavailable. - -Once boundary enforcement is enabled, later plans may be applied while the job -runs. Apply one plan at a time, only advance the boundary, and keep the data topic -and its partition count unchanged during cleanup. - -{{< hint warning >}} -Keep the same cleanup control topic configured and retain its committed records -for every subsequent run and restore of this recovery history. Omitting or -changing the topic removes boundary enforcement; with it configured, the job also -rejects a fresh start without a recovery marker. -{{< /hint >}} - {{< details title="Command examples" >}} Run the tool where the recovery point and Kafka are accessible. The classpath @@ -309,6 +294,21 @@ A successful apply prints `Cleanup plan is APPLIED`. {{< /details >}} +If the first cleanup fails or is interrupted, keep the job stopped and retry +`apply` with the same plan: the boundary may already have been committed, making +earlier recovery points unavailable. + +Once boundary enforcement is enabled, later plans may be applied while the job +runs. Apply one plan at a time, only advance the boundary, and keep the data topic +and its partition count unchanged during cleanup. + +{{< hint warning >}} +Keep the same cleanup control topic configured and retain its committed records +for every subsequent run and restore of this recovery history. Omitting or +changing the topic removes boundary enforcement; with it configured, the job also +rejects a fresh start without a recovery marker. +{{< /hint >}} + #### Switching from per-key tombstones To switch a job from per-key tombstones to checkpoint-aligned cleanup: From a615e766217072a2c817954baae5103bf309b154 Mon Sep 17 00:00:00 2001 From: joeyutong Date: Tue, 20 Oct 2026 19:18:50 +0800 Subject: [PATCH 12/13] [docs] Address consistency guide review feedback Keep current recovery requirements in the recovery section and isolate the known 0.4 format change in version-specific migration notes. Consolidate Kafka cleanup prerequisites, topic requirements, and CLI guidance in Consistency. Generated-by: Codex CLI 0.163.0-alpha.10.2 (GPT-6) Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 105/105 AI-Contributed/UT: 0/0 --- docs/content/docs/operations/configuration.md | 33 +-------- docs/content/docs/operations/consistency.md | 72 ++++++++++--------- 2 files changed, 38 insertions(+), 67 deletions(-) diff --git a/docs/content/docs/operations/configuration.md b/docs/content/docs/operations/configuration.md index cc15a5d4a..1fd208f5f 100644 --- a/docs/content/docs/operations/configuration.md +++ b/docs/content/docs/operations/configuration.md @@ -274,7 +274,7 @@ for passing Kafka client SASL/SSL settings. | Key | Default | Type | Description | |-------------------------------------|--------------------------|---------|-----------------------------------------------------------------------------| -| `kafkaActionStateCleanupControlTopic` | (none) | String | Separate topic containing committed checkpoint-aligned cleanup boundaries. When configured, recovery rejects points requiring data below the boundary. Mutually exclusive with `kafkaActionStateTombstoneEnabled=true`. See [Checkpoint-aligned Kafka cleanup](#checkpoint-aligned-kafka-cleanup) for prerequisites. | +| `kafkaActionStateCleanupControlTopic` | (none) | String | Separate topic containing committed checkpoint-aligned cleanup boundaries. When configured, recovery rejects points requiring data below the boundary. Mutually exclusive with `kafkaActionStateTombstoneEnabled=true`. See [Checkpoint-aligned Kafka cleanup]({{< ref "docs/operations/consistency#checkpoint-aligned-kafka-cleanup" >}}) for prerequisites and procedure. | | `kafkaActionStateTombstoneEnabled` | false | Boolean | Write deletion markers for execution records covered by completed checkpoints. Enabling this may cause completed Actions to run again when restoring older checkpoints or savepoints. Mutually exclusive with `kafkaActionStateCleanupControlTopic`. See [Kafka retention and cleanup]({{< ref "docs/operations/consistency#kafka-retention-and-cleanup" >}}) for the recovery trade-off. | | `kafkaActionStateTopic` | (none) | String | Topic storing Action state. Required when `actionStateStoreBackend` is `kafka`. | | `kafkaActionStateTopicNumPartitions` | 64 | Integer | Number of partitions when creating the topic. Does not change an existing topic. | @@ -292,37 +292,6 @@ When reusing checkpoints or savepoints with this store: - Retention changes and manual deletion must preserve the records needed by every recovery point you intend to retain. See [Kafka retention and cleanup]({{< ref "docs/operations/consistency" >}}#kafka-retention-and-cleanup). -- Once a checkpoint-aligned cleanup boundary has been committed, keep the same - `kafkaActionStateCleanupControlTopic` configuration and its committed history - for all subsequent runs and restores. - -##### Checkpoint-aligned Kafka cleanup - -`KafkaActionStateCleanupTool` requires the following configuration and recovery -constraints: - -- The data topic must use `cleanup.policy=compact,delete`, `retention.ms=-1`, and - `retention.bytes=-1`. These are also the settings used when the store creates a - new data topic. -- The control topic must differ from the data topic, have exactly one partition, - and use `cleanup.policy=compact` without delete retention. Dedicate it to the - same logical Agent operator and physical data topic; do not share it across - operators. -- Before enabling `kafkaActionStateCleanupControlTopic` on the job, the control - topic must exist and contain a committed cleanup boundary. A missing topic or - committed history prevents recovery. -- Per-key tombstones must be disabled. Cleanup boundaries can move only forward; - run only one `apply` operation at a time. -- The `plan` command accepts only versioned Kafka recovery markers. Legacy map - markers can still be used for job recovery, but cannot authorize cleanup. - -The `apply` command's optional `--replication-factor` controls creation of the -control topic. It defaults to `1` and accepts values from `1` to `32767`. - -For the procedure and command examples, see -[Cleanup steps]({{< ref "docs/operations/consistency" >}}#cleanup-steps). Jobs that -previously emitted tombstones must first follow -[Switching from per-key tombstones]({{< ref "docs/operations/consistency" >}}#switching-from-per-key-tombstones). #### Fluss-based Action State Store diff --git a/docs/content/docs/operations/consistency.md b/docs/content/docs/operations/consistency.md index f77f92d4e..e77641a2c 100644 --- a/docs/content/docs/operations/consistency.md +++ b/docs/content/docs/operations/consistency.md @@ -186,9 +186,13 @@ To match saved records to the correct work, replay must preserve the following: Action must reach the same durable calls in the same order, with the same inputs. See [Durable Execution]({{< ref "docs/development/create-agent/actions#durable-execution" >}}) for call identity, serialization, and Action-context requirements. - -Before restoring state after changes to keys, Action names, result types, -application code, or versions, review [Upgrades and migration](#upgrades-and-migration). +- **Keep keys and Action names stable.** Reuse the original key types and + serializer configuration, and ensure custom key serializers produce + deterministic bytes. Changing key serialization or renaming an Action can + prevent recovery from finding its saved completion. +- **Keep saved values usable.** The application must still be able to read and use + stored Events, Memory updates, and durable call outcomes, since recovery may + replay them without running the code that produced them. ### Limits on external side effects @@ -243,10 +247,28 @@ finished. For per-key tombstones, set `kafkaActionStateTombstoneEnabled=true` before starting the job. -For checkpoint-aligned cleanup, use `KafkaActionStateCleanupTool` for both Java -and Python jobs. Review the [configuration prerequisites]({{< ref "docs/operations/configuration#checkpoint-aligned-kafka-cleanup" >}}) -before running the tool. If the job has used per-key tombstones, first follow -[Switching from per-key tombstones](#switching-from-per-key-tombstones). +##### Checkpoint-aligned Kafka cleanup + +Use `KafkaActionStateCleanupTool` for both Java and Python jobs. Before generating +a plan, check the prerequisites below. If the job has used per-key tombstones, +first follow [Switching from per-key tombstones](#switching-from-per-key-tombstones). + +{{< details title="Prerequisites and topic settings" >}} + +- The data topic must use `cleanup.policy=compact,delete`, `retention.ms=-1`, and + `retention.bytes=-1`. These are also the settings used when the store creates a + new data topic. +- The control topic must differ from the data topic, have exactly one partition, + and use `cleanup.policy=compact` without delete retention. Dedicate it to the + same logical Agent operator and physical data topic; do not share it across + operators. +- Before enabling `kafkaActionStateCleanupControlTopic` on the job, the control + topic must exist and contain a committed cleanup boundary. Recovery fails if + the topic or its committed history is missing. +- The `plan` command accepts only versioned Kafka recovery markers. Legacy map + markers can still be used for job recovery, but cannot authorize cleanup. + +{{< /details >}} For the first cleanup of an existing job: @@ -279,9 +301,8 @@ java -cp "$FLINK_HOME/lib/*::}}). +After checking the plan, apply it with a separate control topic. The tool creates +the topic if it does not exist. ```bash java -cp "$FLINK_HOME/lib/*::" \ @@ -290,6 +311,10 @@ java -cp "$FLINK_HOME/lib/*:: is APPLIED`. {{< /details >}} @@ -330,32 +355,9 @@ Flink Agents currently provides no equivalent of Kafka's per-key tombstones or checkpoint-aligned cleanup for the Fluss Action State Store. Instead, use Fluss's own log retention settings to manage record retention and cleanup. -## Upgrades and migration - -### Check recovery conditions before a change - -Before restoring state after a change, check Flink's state compatibility -requirements and confirm that the application can still use its saved Action -results. A readable checkpoint does not by itself guarantee that saved results -remain valid for the new code. - -| Planned change | What to check | -| --- | --- | -| Flink Agents or Flink version | Follow the target release's migration guidance and Flink's upgrade requirements. A job starting successfully does not confirm that recovery across those versions is supported. | -| Key type or serializer | Keep the original key type and serializer configuration when reusing Action state. Custom key serializers must produce deterministic bytes. Changes to the serialized key bytes are not always detected. | -| Action name | Renaming an Action prevents recovery from finding its saved completion. Keeping the same name allows recovery to reuse saved output without running the new body. Check that this output is still valid for the new code. | -| User code, Events, Memory, or durable results | Check that saved outputs and values can still be read and used, and that unfinished Actions reach the same logical calls. Runtime format checks do not verify compatibility with the application's logic. | -| Action State Store backend | When reusing checkpoints or savepoints, preserve the original [Kafka topic]({{< ref "docs/operations/configuration" >}}#changes-to-an-existing-kafka-store) or [Fluss table]({{< ref "docs/operations/configuration" >}}#changes-to-an-existing-fluss-store), its partition or bucket layout, and [the records needed for recovery](#action-state-store-retention-and-cleanup). | - -Use Flink's -[application and version upgrade guidance](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/ops/upgrading/) -for the hosting job. The checks above identify recovery requirements and known -risks. They do not establish which Flink Agents and Flink version combinations are -supported. - -### Migration notes +## Migration notes -#### Flink Agents 0.4 +### Flink Agents 0.4 Flink Agents 0.4 changes the Action State Store record key format and does not support records written by earlier versions. When upgrading from an earlier From 484a58b5e3d39de4eb9310636e56c0d862efeb6a Mon Sep 17 00:00:00 2001 From: joeyutong Date: Tue, 20 Oct 2026 20:49:16 +0800 Subject: [PATCH 13/13] [docs] Remove migration guidance from Consistency Keep the guide focused on current consistency and recovery behavior. Remove the 0.4 migration notes and align Deployment's link description. Generated-by: Codex CLI 0.163.0-alpha.10.2 (GPT-6) Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 11/11 AI-Contributed/UT: 0/0 --- docs/content/docs/operations/consistency.md | 9 --------- docs/content/docs/operations/deployment.md | 2 +- 2 files changed, 1 insertion(+), 10 deletions(-) diff --git a/docs/content/docs/operations/consistency.md b/docs/content/docs/operations/consistency.md index e77641a2c..07a6a312a 100644 --- a/docs/content/docs/operations/consistency.md +++ b/docs/content/docs/operations/consistency.md @@ -354,12 +354,3 @@ affected by past tombstones. Flink Agents currently provides no equivalent of Kafka's per-key tombstones or checkpoint-aligned cleanup for the Fluss Action State Store. Instead, use Fluss's own log retention settings to manage record retention and cleanup. - -## Migration notes - -### Flink Agents 0.4 - -Flink Agents 0.4 changes the Action State Store record key format and does not -support records written by earlier versions. When upgrading from an earlier -version, use a new Kafka topic or Fluss table and submit the new job without -restoring a checkpoint or savepoint from the previous version. diff --git a/docs/content/docs/operations/deployment.md b/docs/content/docs/operations/deployment.md index de81ba1c0..46aad7c3e 100644 --- a/docs/content/docs/operations/deployment.md +++ b/docs/content/docs/operations/deployment.md @@ -91,4 +91,4 @@ Now you should see a Flink job submitted to the Flink Cluster in Flink web UI (t Before submitting the job, choose and configure its consistency mode. See [Consistency]({{< ref "docs/operations/consistency" >}}) for required setup, -recovery guarantees, retention and cleanup, and upgrade guidance. +recovery guarantees, retention and cleanup.