wenjin272 commented on code in PR #1284: URL: https://github.com/apache/flink-agents/pull/1284#discussion_r4322580150
########## docs/content/docs/operations/consistency.md: ########## @@ -0,0 +1,363 @@ +--- +title: Consistency +weight: 2 +type: docs +--- +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +When a Flink job recovers from a checkpoint, it restores its state and replays the +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 + +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**. + +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 | +| --- | --- | --- | +| No checkpointing | The job restarts from the source's configured starting positions without restoring Agent state or progress from a checkpoint. | Suitable when the application can accept lost progress or repeated work. | +| Flink checkpointing | Flink restores checkpointed state and replays subsequent input. Actions and external calls may repeat, and models or queries may return different results. | Enable exactly-once checkpointing with durable checkpoint storage and the required source and sink guarantees. External calls must tolerate repetition. | +| Flink checkpointing + Action State Store | Recovery replays saved completions without rerunning Action bodies. Unfinished Actions may run again, reusing saved durable call outcomes. | Add a Kafka or Fluss Action State Store. Saving Action completions and durable call outcomes requires synchronous backend writes. Retain both checkpoints and the backend records needed for recovery. | + +## Required setup + +Enable exactly-once checkpointing for the Flink or PyFlink job, and use source +and sink connectors with 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 + +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. + +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 + +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 >}} + +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). + +#### 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 needed. By default, +it uses the `flink_agents` database and creates new tables with 64 buckets. + +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 + +### Recovery flow + +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, 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. + - **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 + +To match saved records to the correct work, replay must preserve the following: + +- **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.** An unfinished + 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). + +### Limits on external side effects + +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. +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 may need to restore, retain both the +snapshot and the Action State Store records it requires. Before changing retention +or cleanup settings, identify the oldest recovery point you need 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. + +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, +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 | +| --- | --- | --- | --- | +| 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. | Cache cleanup alone does not invalidate 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 no longer needed by the selected recovery point. | You need to retain a selected checkpoint or savepoint while reclaiming space from older recovery data. | 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, 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). + +For the first cleanup of an existing job: + +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**. + +{{< 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-<FLINK_VERSION>-<FA_VERSION>.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/*:<FLINK_AGENTS_DIST_JAR>:<FLINK_STATE_PROCESSOR_API_JAR>" \ + org.apache.flink.agents.runtime.actionstate.KafkaActionStateCleanupTool \ + plan --checkpoint "<CHECKPOINT_OR_SAVEPOINT_PATH>" \ + --operator-uid-hash "<AGENT_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/*:<FLINK_AGENTS_DIST_JAR>:<FLINK_STATE_PROCESSOR_API_JAR>" \ + 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 <id> 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: + +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 + +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 Review Comment: I think we can remove the “Upgrades and migration” section from this guide. We plan to design compatibility for the Action State Store and Flink operator state in the next release. For now, the guide should describe current recovery requirements without implying broader upgrade compatibility. The existing 0.4 format change can go into release-specific migration notes. Please also remove the reference to #1055, which is already closed. ########## docs/content/docs/operations/configuration.md: ########## @@ -103,261 +96,265 @@ agent: #### Loading Behavior -By default, the configuration is automatically loaded from `$FLINK_HOME/conf/config.yaml`. - -**Special Condition** +When creating an `AgentsExecutionEnvironment` for a Flink job, Flink Agents loads +the `agent` section from the configuration directory specified by `FLINK_CONF_DIR`. +Set this environment variable before creating the environment: -In the following case, Flink Agents may not locate the corresponding configuration file, necessitating manual configuration. If the file is not set, no configuration file will be loaded, potentially resulting in unexpected behavior or failures. - -- **For MiniCluster**: - Manual setup is **required** — always export the environment variable before running the job: +```bash +export FLINK_CONF_DIR="/path/to/flink/conf" +``` - ```bash - export FLINK_CONF_DIR="path/to/your/config.yaml" - ``` +`FLINK_CONF_DIR` points to a directory. Java reads `config.yaml` from that directory. +Python reads `flink-conf.yaml` if it exists, otherwise `config.yaml`. - This ensures that Flink can locate and load the configuration file correctly. +If `FLINK_CONF_DIR` is unset, no YAML configuration file is loaded. Configuration +through `AgentsExecutionEnvironment` still works and overrides values loaded from +the file. ## Built-in configuration options -### Core Options -Here is the list of all built-in core configuration options. - -| Key | Default | Type | Description | -|---------------------------|----------------------------|-----------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| `event-listeners` | none | `List<String>` | The list of event listener class names. Each class must implement the EventListener interface and provide a public no-argument constructor. {{< hint warning >}} Note: Currently, custom event listeners are only supported in Java. {{< /hint >}} | -| `action.trigger-condition.evaluate-failure-strategy` | `WARN_AND_SKIP` | ConditionEvaluationFailureStrategy | Handles failures while evaluating a trigger condition for an individual Event, including missing attributes, incompatible values, or a non-Boolean result. <br/><ul><li>`WARN_AND_SKIP` (default): log a warning, treat that condition as false, and continue with later OR conditions.</li><li>`FAIL`: throw `IllegalStateException` and fail the Flink task; recovery follows the job's restart configuration.</li></ul> Plan-validation failures and runtime compilation or static type-check failures occur during initialization and are not handled by this option. | -| `max-retries` | 0 | int | Number of additional attempts per model call, including routing judge calls. Defaults to 0 (no retries). | -| `retry-wait-interval` | 1 | int | Base wait interval in seconds between retries. Uses exponential backoff: the actual wait time for the Nth retry is `retry-wait-interval * 2^(N-1)` seconds. For example, with default 1s, waits are 1s, 2s, 4s, etc. The built-in chat and decision actions report retry count and total wait time in `ChatResponseEvent` and `DecisionResponseEvent`, respectively, and record metrics (`retryCount`, `retryWaitSec`) under the configured model resource name. | -| `chat.async` | true | boolean | Whether chat asynchronously for built-in chat action. | -| `decision.async` | true | boolean | Whether the built-in decision action invokes the model through durable async execution. Java calls through Python fall back to synchronous execution when the Flink runtime does not support asynchronous Python calls. | -| `tool-call.async` | true | boolean | Whether the built-in tool-call action runs each tool via durable async execution. See [Parallel tool-call batches]({{< ref "docs/development/calling-llms/tool_use#parallel-tool-call-batches" >}}) for how this switch works with the batch settings below. | -| `async.batch.parallelism` | os cpu count | int | Maximum concurrent calls in one async batch, including batches created with `ctx.gather(...)`. `1` runs calls sequentially. Actual concurrency is also limited by `num-async-threads`. See [Parallel tool-call batches]({{< ref "docs/development/calling-llms/tool_use#parallel-tool-call-batches" >}}). | -| `async.batch.timeout.ms` | -1 (disabled) | long (milliseconds) | Overall timeout for one async batch. Non-positive values disable it. A timeout does not guarantee that running calls stop or undo their external effects. See [Timeouts and external effects]({{< ref "docs/development/calling-llms/tool_use#timeouts-and-external-effects" >}}) for tool-call behavior and constraints. | -| `rag.async` | true | boolean | Whether retrieve context asynchronously for built-in context retrieval action. | -| `num-async-threads` | os cpu count * 2 | int | Controls execution capacity for asynchronous calls. `async.batch.parallelism` separately limits concurrency within each batch. See [Parallel tool-call batches]({{< ref "docs/development/calling-llms/tool_use#parallel-tool-call-batches" >}}) for tool-call concurrency guidance. | -| `parallel-execution.enabled` | true | boolean | Whether plans with only **Java Actions** run Actions concurrently on the parallel execution engine when the continuation-based engine is unavailable (**JDK < 21**). Plans declaring Python Resources additionally require a Flink version supporting [cross-language async calls]({{< ref "docs/development/create-agent/actions#cross-language-async-requirements" >}}). Same-key input records commit in input order, with exactly-once checkpoint semantics. Set to `false` to use the serial engine. Ignored when the JDK 21 coroutine engine is available, and for plans containing **Python** actions, which never use the parallel engine. {{< hint warning >}}**Experimental.** If you observe unexpected behavior on JDK < 21, set this option to `false` and report an issue.{{< /hint >}} | -| `max-in-flight-input-records` | 100 | int | Maximum number of input records that may be in flight concurrently. Only enforced by the **JDK < 21** parallel execution engine for plans with only **Java Actions**; the JDK 21 coroutine engine and plans containing **Python** actions ignore it. Every admitted record consumes one unit of budget; at the cap, admission of further records blocks until an in-flight record retires. | -| `job-identifier` | none | String | The unique identifier of job, remaining consistent after restoring from a savepoint. If not set, uses flink job id. | -| `short-term-memory.state-ttl.ms` | 0 | long | Time-to-live for short-term memory state in milliseconds. Set to a value greater than 0 to enable TTL; 0 disables it. | -| `short-term-memory.state-ttl.update-type` | `ON_READ_AND_WRITE` | ShortTermMemoryTtlUpdate | Update policy for short-term memory TTL. Only applies when `short-term-memory.state-ttl.ms` is greater than 0. Valid values: `ON_CREATE_AND_WRITE`, `ON_READ_AND_WRITE`. An enabled run-begin memory snapshot also refreshes TTL for entries it reads under `ON_READ_AND_WRITE`. | -| `short-term-memory.state-ttl.visibility` | `NEVER_RETURN_EXPIRED` | ShortTermMemoryTtlVisibility | Visibility policy for expired short-term memory state. Only applies when `short-term-memory.state-ttl.ms` is greater than 0. Valid values: `NEVER_RETURN_EXPIRED`, `RETURN_EXPIRED_IF_NOT_CLEANED_UP`. | - -### Trace Log Options - -Trace Log records Events, Action executions, and calls made within Actions. See [Monitoring]({{< ref "docs/operations/monitoring#trace-log" >}}) for output formats and the Trace Tree reader. - -| Key | Default | Type | Description | -|-----|---------|------|-------------| -| `trace-log.output-type` | `SLF4J` | LoggerType | Trace log output: `SLF4J` writes JSON through a dedicated SLF4J logger, visible in Flink's Web UI Logs tab; `FILE` writes per-subtask files. A non-empty `trace-log.base-dir` forces `FILE`, even when `SLF4J` is selected. | -| `trace-log.base-dir` | (none) | String | Directory for file output; a non-empty value forces `FILE`. When file output is selected without a directory, uses `java.io.tmpdir/flink-agents`. | -| `trace-log.pretty-print` | false | boolean | Format JSON over multiple lines. When false, each record occupies one line (JSONL). | -| `trace-log.targets` | `[{scope: EVENT_ONLY, level: STANDARD}]` | List of objects | Match records and control the recording level. Each target uses `scope` and optional `level`; see [Trace log targets](#trace-log-targets). | -| `trace-log.standard.max-string-length` | 2000 | int | Maximum attribute string length at `STANDARD` level. No effect at `VERBOSE`. | -| `trace-log.standard.max-array-elements` | 20 | int | Maximum attribute array length at `STANDARD` level. No effect at `VERBOSE`. | -| `trace-log.standard.max-depth` | 5 | int | Maximum attribute nesting depth at `STANDARD` level. Deeper objects are summarized. No effect at `VERBOSE`. | - -#### Trace log targets - -Every target combines a `scope` with an optional `level`. Omitting `targets` records only Events at `STANDARD`. An explicit list replaces that default; an empty list records nothing. +<span id="core-options"></span> -For example, this configuration records all entities at `STANDARD`, with full attributes for the `search` Tool: +### Observability -```yaml -agent: - trace-log: - targets: - - scope: ALL - level: STANDARD - - scope: {tool: search} - level: VERBOSE - standard: - max-string-length: 2000 - max-array-elements: 20 - max-depth: 5 - output-type: FILE - base-dir: /tmp/flink-agent-logs - pretty-print: false -``` +See [Monitoring]({{< ref "docs/operations/monitoring" >}}) for Trace Log output and +record selection, and [Memory Observability]({{< ref "docs/development/memory/memory_events" >}}) +for the agent-run begin event and custom Event listeners. -Java and Python can configure targets with `TraceLogTarget` through the same `TRACE_LOG_TARGETS` option: +<span id="trace-log-options"></span> -{{< tabs >}} -{{< tab "python" >}} - -```python -from flink_agents.api.core_options import AgentConfigOptions -from flink_agents.api.trace_log import TraceLogLevel, TraceLogTarget - -agents_env.get_config().set( - AgentConfigOptions.TRACE_LOG_TARGETS, - [ - TraceLogTarget.all(TraceLogLevel.STANDARD), - TraceLogTarget.tool("search", TraceLogLevel.VERBOSE), - ], -) -``` - -{{< /tab >}} -{{< tab "java" >}} - -```java -import java.util.List; -import org.apache.flink.agents.api.configuration.AgentConfigOptions; -import org.apache.flink.agents.api.logger.TraceLogLevel; -import org.apache.flink.agents.api.logger.TraceLogTarget; - -agentsEnv.getConfig().set( - AgentConfigOptions.TRACE_LOG_TARGETS, - List.of( - TraceLogTarget.all(TraceLogLevel.STANDARD), - TraceLogTarget.tool("search", TraceLogLevel.VERBOSE))); -``` - -{{< /tab >}} -{{< /tabs >}} +| Key | Default | Type | Description | +| --- | --- | --- | --- | +| `agent-run.begin-event` | false | boolean | Emit an `AgentRunBeginEvent` with a short-term memory value snapshot before running the Actions triggered by each input. Independent of `memory.generate-event`. | +| `event-listeners` | none | `List<String>` | Fully qualified Java `EventListener` class names. Each class must provide a public no-argument constructor. Python listener implementations are not supported. | +| `trace-log.base-dir` | (none) | String | Directory for Trace Log files. A nonblank value selects `FILE`, overriding `trace-log.output-type`. When unset, file output uses `${java.io.tmpdir}/flink-agents`. | +| `trace-log.output-type` | `SLF4J` | LoggerType | Trace Log destination: `SLF4J` uses the Trace Log logger; `FILE` writes one file per operator subtask. See [Output]({{< ref "docs/operations/monitoring#output" >}}) for logging configuration and file locations. | +| `trace-log.pretty-print` | false | boolean | Write multiline JSON when `true`, or compact single-line JSON when `false`. Applies to both output types. | +| `trace-log.standard.max-array-elements` | 20 | int | Maximum elements retained in each attribute array at `STANDARD` level. `0` disables this limit. No effect at `VERBOSE`. | +| `trace-log.standard.max-depth` | 5 | int | Nesting depth at which attribute objects are summarized at `STANDARD` level. `0` disables this limit. No effect at `VERBOSE`. | +| `trace-log.standard.max-string-length` | 2000 | int | Maximum characters retained in each attribute string at `STANDARD` level. `0` disables this limit. No effect at `VERBOSE`. | +| `trace-log.targets` | `[{scope: EVENT_ONLY, level: STANDARD}]` | List of objects | Select Trace Log entities and their recording levels. Each target requires `scope` and may specify `level`; see [Trace log targets](#trace-log-targets). | -`TraceLogTarget` provides `all()` and whole-type factories `events()`, `actions()`, `tools()`, `llms()`, `parsers()`, and `subagents()`. The corresponding singular factories accept a name, such as `event("OrderCreated")` or `tool("search")`. All factories accept an optional `TraceLogLevel`. Omitting it preserves inheritance: `tool("search")` inherits the level of a matching Tool type target or `ALL`, as described below. Targets are immutable. +#### Trace log targets -In YAML, `scope` accepts `EVENT_ONLY` (match Events), `ALL` (match all entities), or a selector object. Use one of the keys `event`, `action`, `tool`, `llm`, `parser`, or `subagent` with an entity name, for example `{event: OrderCreated}`. The value `"*"` selects that whole type: `{event: "*"}` is equivalent to `EVENT_ONLY`. +`trace-log.targets` is a list of target objects. Omitting the option records Events +at `STANDARD`. An explicit list replaces this default; an empty list records nothing. -The generic form `{entityType: custom, entityName: name}` also supports custom entity types reported through `ExecutionReporter`; omit `entityName` to select the whole type. Use `TraceLogTarget.entity(...)` for the same selection in Java or Python. For example, `TraceLogTarget.entity("retriever", "catalog")` corresponds to `scope: {entityType: retriever, entityName: catalog}`. Each selector object must contain exactly one category key, or the generic `entityType` and optional `entityName` fields. Multiple category keys and mixing a category key with the generic fields are rejected. +Each target has the following fields: -Direct model or tool calls inside an Action are not automatically recorded as separate calls, even with `scope: ALL`. +- `scope` (required): `EVENT_ONLY`, `ALL`, a category selector with exactly one of + `event`, `action`, `tool`, `llm`, `parser`, or `subagent`, or a generic selector with + `entityType` and optional `entityName`. Do not mix category and generic fields. + A category value of `"*"`, or a generic selector without `entityName`, selects the + whole entity type. `EVENT_ONLY` is equivalent to `{event: "*"}`. +- `level` (optional): `OFF`, `STANDARD`, or `VERBOSE`. An omitted level inherits + from matching scopes; it does not automatically mean `STANDARD`. In YAML, quote + `level: "OFF"` so Python does not interpret it as a boolean. -`entityType` identifies the kind of object to observe. For `entityName`, use the logical name from your Agent's code or configuration. Trace Log records and target selectors use the same name. Common examples are: +Duplicate scopes may combine an omitted level with an explicit level or repeat the +same explicit level. Conflicting explicit levels for the same scope are rejected +when logging starts. This also applies to equivalent Event type selectors. -| `entityType` | What to use as `entityName` | Example | -|--------------|----------------------------|---------| -| `event` | The Event type used when emitting or subscribing to Events. | `OrderCreated` | -| `action` | The name used to declare the Action in the Agent. | `create_order` | -| `llm` | The registered ChatModel Resource name used by the call. | `primary_model` | +For YAML and API examples, entity naming, and level inheritance, see +[Recording Targets]({{< ref "docs/operations/monitoring#recording-targets" >}}). -For example, `scope: {llm: primary_model}` selects calls to the Resource named `primary_model`. Its configured model or deployment is recorded separately in `entityMetadata.model`. +The old `event-log.*`, `eventLoggerType`, `baseLogDir`, and `prettyPrint` keys are +unsupported and rejected at startup. Use `trace-log.targets`, `trace-log.standard.*`, +and the output options above. -Entity names match exactly. A name ending in `.*` matches its namespace and descendants: `{event: "com.foo.*"}` matches `com.foo` and `com.foo.Event`, but not `com.foobar.Event`. Other wildcard forms are unsupported. A category selector with `"*"`, or a generic selector without `entityName`, matches the whole type. +### Condition handling -`level` accepts `OFF`, `STANDARD`, and `VERBOSE`. `OFF` writes no record. `STANDARD` truncates or summarizes attributes using the `standard` limits; `VERBOSE` retains full attributes. Both recording settings preserve all identity and relationship fields, entity metadata, timestamps, statuses, and problem categories. Media payload sanitization applies at both recording settings. See [Recording Level]({{< ref "docs/operations/monitoring#recording-level" >}}) for the truncation format and media handling. In YAML, write `level: "OFF"` with quotes so Python does not interpret it as a boolean. +See [Condition Evaluation Failures]({{< ref "docs/development/create-agent/actions#condition-evaluation-failures" >}}) +for trigger condition examples. -For each entity, matching scopes are considered from most to least specific: +| Key | Default | Type | Description | +| --- | --- | --- | --- | +| `action.trigger-condition.evaluate-failure-strategy` | `WARN_AND_SKIP` | ConditionEvaluationFailureStrategy | Behavior when evaluating a trigger condition for an Event fails. `WARN_AND_SKIP` logs a warning and treats the condition as false; `FAIL` fails the Flink task. Initialization failures are not affected. See [Condition Evaluation Failures]({{< ref "docs/development/create-agent/actions#condition-evaluation-failures" >}}). | -| Order | Matching scope | -|-------|----------------| -| 1 | Exact entity name within its type | -| 2 | Namespace prefixes within its type, from longest to shortest | -| 3 | Entity type (`EVENT_ONLY` is the `event` type scope) | -| 4 | `ALL` | +### Model calls, tools, and retries -The first explicitly specified `level` in this chain applies. An omitted `level` inherits from the next matching scope and does not override it. If at least one scope matches but no matching scope specifies `level`, the entity uses `STANDARD`. If no scope matches, no record is written. Target order does not affect the result. +See [Chat]({{< ref "docs/development/calling-llms/chat#handle-success-and-failure" >}}) +and [Decision Models]({{< ref "docs/development/decision_models#event-driven-calls" >}}) +for model-call outcomes and retries, and [Context Retrieval]({{< ref "docs/development/vector-search/vector_stores#built-in-events-and-actions" >}}) +for vector-store queries. See [Tool Use]({{< ref "docs/development/calling-llms/tool_use#parallel-tool-call-batches" >}}) +for tool-call batches and timeout behavior, and [Execution Metrics]({{< ref "docs/operations/monitoring#execution-metrics" >}}) +for retry metrics. -An explicit `OFF` stops inheritance and suppresses recording. A more specific scope can enable recording by explicitly setting `STANDARD` or `VERBOSE`; omitting `level` continues to inherit `OFF`. +| Key | Default | Type | Description | +| --- | --- | --- | --- | +| `chat.async` | true | boolean | Enable durable async execution for built-in Chat model calls, including routing judge calls. `false` selects synchronous execution. | +| `decision.async` | true | boolean | Enable durable async execution for built-in Decision Model calls. `false` selects synchronous execution. | +| `max-retries` | 0 | int | Maximum retries after the initial attempt of a built-in Chat or Decision Model call, including routing judge calls. Non-positive values disable retries. | +| `rag.async` | true | boolean | Enable durable async execution for vector-store queries made by the built-in context retrieval Action. `false` selects synchronous execution. | +| `retry-wait-interval` | 1 | int | Base wait in seconds between retries controlled by `max-retries`. The wait before retry N is `retry-wait-interval * 2^(N-1)`. Non-positive values retry without waiting. | +| `tool-call.async` | true | boolean | Enable durable async execution for ordinary tools handled by the built-in Tool Call Action. With `false`, these calls run synchronously, one at a time. See [Tool Use]({{< ref "docs/development/calling-llms/tool_use#parallel-tool-call-batches" >}}) for batch behavior. | + +These async switches select how built-in calls execute. Synchronous calls still +use durable execution; saving outcomes for recovery requires an +[Action State Store]({{< ref "docs/operations/consistency#configure-an-action-state-store" >}}). +Built-in Chat and context retrieval calls, and Java Decision Model calls through +Python, fall back to synchronous execution when the runtime cannot support their +cross-language async call. See [Execution Requirements]({{< ref "docs/development/create-agent/actions#execution-requirements" >}}) +for cross-language async requirements. + +The `ErrorHandlingStrategy` API and `error-handling-strategy` option have been +removed. Configure retries with `max-retries`; see Chat for terminal failure handling. + +### Runtime and async execution + +See [Execution Requirements]({{< ref "docs/development/create-agent/actions#execution-requirements" >}}) +for JDK, scheduling, and cross-language requirements. Batch limits also apply to +the built-in [parallel tool-call batches]({{< ref "docs/development/calling-llms/tool_use#parallel-tool-call-batches" >}}). -For example, a Tool type target at `VERBOSE` supplies the level to matching namespace and exact-name targets that omit it. A matching namespace target at `OFF` suppresses an exact-name target that omits `level`, while an exact-name target with explicit `STANDARD` or `VERBOSE` enables that entity. `EVENT_ONLY` applies only to Events; its level does not affect a Tool target. +| Key | Default | Type | Description | +| --- | --- | --- | --- | +| `async.batch.parallelism` | os cpu count | int | Maximum concurrent calls in one `ctx.gather(...)` batch, including ordinary, durable, and built-in tool calls. Values of `1` or less run calls sequentially. Worker-pool capacity can further limit concurrency. | +| `async.batch.timeout.ms` | -1 (disabled) | long (milliseconds) | Overall timeout for one `ctx.gather(...)` batch, in milliseconds. Non-positive values disable it. A timeout does not guarantee that running calls stop and cannot undo their external effects. See [Tool Use]({{< ref "docs/development/calling-llms/tool_use#timeouts-and-external-effects" >}}) for built-in tool-batch behavior. | +| `job-identifier` | none | String | Stable job identifier used to scope Mem0 long-term memory. Keep an explicit value unchanged across restores. If unset, the initial Flink JobID is stored in operator state and reused when restoring a checkpoint or savepoint. | +| `max-in-flight-input-records` | 100 | int | Maximum input records in flight per Agent operator subtask on the experimental parallel engine. Must be positive. The limit is ignored by continuation-based execution and plans containing Python Actions. See [Execution Requirements]({{< ref "docs/development/create-agent/actions#execution-requirements" >}}). | +| `num-async-threads` | os cpu count * 2 | int | Worker threads per async execution pool on each Agent operator subtask. Must be positive. `async.batch.parallelism` separately limits concurrency within each batch; see [Execution Requirements]({{< ref "docs/development/create-agent/actions#execution-requirements" >}}) for scheduling behavior. | +| `parallel-execution.enabled` | true | boolean | **Experimental.** Enable the parallel engine for eligible plans with only Java Actions when continuation-based execution is unavailable. `false` selects serial execution. Ignored by the continuation engine and plans containing Python Actions. See [Execution Requirements]({{< ref "docs/development/create-agent/actions#execution-requirements" >}}) for JDK and Python Resource requirements. | -Duplicate targets for the same scope merge an omitted level with an explicit level and accept repeated identical explicit levels. Different explicit levels for the same scope are rejected when logging starts, regardless of target order. This includes equivalent targets expressed as `EVENT_ONLY`, `{event: "*"}`, or `{entityType: event}`. For example, `EVENT_ONLY` without `level` and `{event: "*"}` with `VERBOSE` merge into one Event type target at `VERBOSE`. Typed targets, mapping objects, and YAML follow the same validation, matching, and inheritance rules. +### Memory -Selecting a Tool does not automatically select its parent Action; recorded relationships retain their IDs even when related records are omitted. +#### Memory events {#memory-event-options} -An `ALL` preset with `level: "OFF"` disables its default range, while a more specific target with explicit `STANDARD` or `VERBOSE` can enable recording. A preset is unnecessary when selecting only one entity type. To record only Tools: +An explicitly configured operation-specific switch overrides `memory.generate-event`. +If neither is set, the effective default below applies. See +[Memory Observability]({{< ref "docs/development/memory/memory_events" >}}) for event +payloads and subscription examples. The independent `agent-run.begin-event` switch +is listed under [Observability](#observability). -```yaml -agent: - trace-log: - targets: - - scope: {tool: "*"} - level: VERBOSE -``` +| Key | Raw default | Effective default | Type | Description | +| --- | --- | --- | --- | --- | +| `memory.generate-event` | unset | per-operation defaults | boolean | Fallback value for operation-specific memory-event switches that are unset. | +| `memory.generate-event.long-term-get` | unset | on | boolean | Emit Events describing long-term memory reads. | +| `memory.generate-event.long-term-search` | unset | on | boolean | Emit Events describing long-term memory searches. | +| `memory.generate-event.long-term-update` | unset | on | boolean | Emit Events describing long-term memory additions, updates, deletions, and memory-set deletion. | +| `memory.generate-event.sensory-read` | unset | off | boolean | Emit Events describing sensory memory reads. | +| `memory.generate-event.sensory-write` | unset | on | boolean | Emit Events describing sensory memory writes. | +| `memory.generate-event.short-term-read` | unset | off | boolean | Emit Events describing short-term memory reads. | +| `memory.generate-event.short-term-write` | unset | on | boolean | Emit Events describing short-term memory writes. | -To record everything except one Tool, use a local `OFF` target: +#### Short-term memory TTL -```yaml -agent: - trace-log: - targets: - - scope: ALL - level: STANDARD - - scope: {tool: search} - level: "OFF" -``` +See [Short-Term Memory TTL]({{< ref "docs/development/memory/sensory_and_short_term_memory#short-term-memory-2" >}}) +for TTL behavior and setup examples. -The old `event-log.*`, `eventLoggerType`, `baseLogDir`, and `prettyPrint` configuration keys are unsupported and rejected at startup. Migrate selection and verbosity to `targets`, truncation limits to `trace-log.standard.*`, and output settings to `trace-log.output-type`, `trace-log.base-dir`, and `trace-log.pretty-print`. +| Key | Default | Type | Description | +| --- | --- | --- | --- | +| `short-term-memory.state-ttl.ms` | 0 | long | Short-term memory TTL in milliseconds. A positive value enables expiration; non-positive values disable it. | +| `short-term-memory.state-ttl.update-type` | `ON_READ_AND_WRITE` | ShortTermMemoryTtlUpdate | When TTL is enabled, refresh expiration on creation and writes (`ON_CREATE_AND_WRITE`), or also on reads (`ON_READ_AND_WRITE`). | +| `short-term-memory.state-ttl.visibility` | `NEVER_RETURN_EXPIRED` | ShortTermMemoryTtlVisibility | When TTL is enabled, never return expired state (`NEVER_RETURN_EXPIRED`), or allow it until cleanup (`RETURN_EXPIRED_IF_NOT_CLEANED_UP`). | -### Memory Event Options +### Action consistency -The eight `memory.generate-event*` options have no raw `ConfigOption` default. When a sub-key and the master switch are both unset, the runtime uses the effective default shown below. See [Memory Observability]({{< ref "docs/development/memory/memory_events" >}}) for resolution order, event payloads, and subscription examples. +<span id="action-state-store"></span> -| Key | Raw default | Effective default | Type | Description | -|-----|-------------|-------------------|------|-------------| -| `memory.generate-event` | unset | per-operation defaults | boolean | Master fallback for unset operation-specific switches. | -| `memory.generate-event.short-term-write` | unset | on | boolean | Emit short-term memory write events. | -| `memory.generate-event.short-term-read` | unset | off | boolean | Emit short-term memory read events. | -| `memory.generate-event.sensory-write` | unset | on | boolean | Emit sensory memory write events. | -| `memory.generate-event.sensory-read` | unset | off | boolean | Emit sensory memory read events. | -| `memory.generate-event.long-term-update` | unset | on | boolean | Emit long-term memory add/delete events. | -| `memory.generate-event.long-term-get` | unset | on | boolean | Emit long-term memory get events. | -| `memory.generate-event.long-term-search` | unset | on | boolean | Emit long-term memory search events. | -| `agent-run.begin-event` | false | off | boolean | Opt in to the agent-run begin event. Independent of the memory-event master switch. | - -### Action State Store +For consistency modes, setup examples, and recovery behavior, see +[Consistency]({{< ref "docs/operations/consistency" >}}). #### Common | Key | Default | Type | Description | |------------------------------|------------------|---------|------------------------------------------------------------------------------------------| -| `actionStateStoreBackend` | (none) | String | The backend for action state store. Supported values: `"kafka"`, `"fluss"`. | +| `actionStateStoreBackend` | (none) | String | Action State Store backend: `"kafka"` or `"fluss"` (case-insensitive). When unset, no external Action State Store is enabled. | #### Kafka-based Action State Store -Here are the configuration options for Kafka-based Action State Store. +Dedicate one topic to each logical Agent operator, shared by that operator's +subtasks. Separate operators in the same job need separate topics. + +The store creates the configured topic if it does not exist, and reuses an existing +topic without changing its settings. The partition-count and replication-factor +options apply only when creating a topic. + +The Kafka Action State Store and its cleanup tool currently expose no configuration +for passing Kafka client SASL/SSL settings. | Key | Default | Type | Description | |-------------------------------------|--------------------------|---------|-----------------------------------------------------------------------------| -| `kafkaBootstrapServers` | "localhost:9092" | String | The config parameter specifies the Kafka bootstrap server. | -| `kafkaActionStateTopic` | (none) | String | The Kafka topic for action state. Dedicate it to one logical Flink Agents operator, shared by that operator's subtasks. | -| `kafkaActionStateTopicNumPartitions`| 64 | Integer | The config parameter specifies the number of partitions for the Kafka action state topic. | -| `kafkaActionStateTopicReplicationFactor` | 1 | Integer | The config parameter specifies the replication factor for the Kafka action state topic. | -| `kafkaActionStateTombstoneEnabled` | false | Boolean | Whether pruning sends tombstone records so log compaction can reclaim pruned keys on a compacted action-state topic. Off by default: pruning does not invalidate older restore points, but the topic continues to grow. When enabled, the checkpoint whose completion triggers pruning remains usable, but restoring an earlier checkpoint or savepoint may replay later tombstones and re-execute already completed actions. Enable only if the job never restores from earlier checkpoints or savepoints, or if re-executing actions is acceptable. | -| `kafkaActionStateCleanupControlTopic` | (none) | String | Separate, single-partition topic containing committed checkpoint-aligned cleanup boundaries. It must use `cleanup.policy=compact` without delete retention, differ from the action-state topic, and be dedicated to the same job recovery history. Setting it enables boundary enforcement during recovery. It cannot be combined with `kafkaActionStateTombstoneEnabled`. | +| `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. | +| `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. | +| `kafkaActionStateTopicReplicationFactor` | 1 | Integer | Replication factor when creating the topic. Does not change an existing topic. | +| `kafkaBootstrapServers` | "localhost:9092" | String | Bootstrap servers for the Kafka Action State Store. | + +##### Changes to an existing Kafka store + +When reusing checkpoints or savepoints with this store: + +- Keep the original topic and partition count. Increasing the partition count + prevents recovery from existing snapshots and causes checkpoints of the running + job to fail. Do not delete and recreate the topic or switch to a replacement, + even if it has the same name and partition count. +- 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 -Checkpoint-aligned cleanup is an explicit administrative operation. `KafkaActionStateCleanupTool` provides the same command-line workflow for Java and Python jobs. Run it with the Flink Agents runtime JAR and the matching Flink State Processor API JAR on the classpath: - -```text -plan --checkpoint PATH (--operator-uid UID | --operator-uid-hash HASH) --output FILE -apply --plan FILE --bootstrap-servers SERVERS --control-topic TOPIC [--replication-factor N] -``` - -The `plan` command reads all recovery markers from the selected checkpoint or savepoint through Flink's State Processor API, creates a deterministic plan using the earliest required offset per partition, and refuses to overwrite an existing output file. Legacy map markers can still be restored by a job, but cannot authorize deletion because they do not identify the physical Kafka topic. The `apply` command verifies the content-derived plan ID before contacting Kafka. Plan files must contain exactly one JSON document; trailing content is rejected. The optional `--replication-factor` controls replication when creating the control topic, defaults to `1`, and must be between `1` and `32767`. - -Review and retain the plan's deterministic JSON before applying it. The coordinator verifies that the selected offsets are still available, writes `COMMITTED` before calling Kafka `deleteRecords`, verifies every resulting beginning offset, and then writes `APPLIED`. Reapplying the same plan retries an interrupted committed operation without changing its boundary. Set `kafkaActionStateCleanupControlTopic` on the job only after the first plan is committed; recovery requires the configured topic to exist and contain a committed boundary, and fails closed if the topic is missing or empty. - -Physical prefix cleanup requires the action-state data topic to use `cleanup.policy=compact,delete`, `retention.ms=-1`, and `retention.bytes=-1`. The store uses those settings when it creates a new topic. Before applying cleanup to an existing topic, alter it to those values. The `delete` policy enables Kafka's `deleteRecords` API, while both retention limits remain disabled so Kafka cannot independently retire records that a supported checkpoint still needs. Apply validates these effective settings before writing `COMMITTED` and rechecks them immediately before and after deletion. - -For the first cleanup on an existing job, stop the job, create and apply the plan from the recovery point that will become the oldest supported one, and restart from that same point with `kafkaActionStateCleanupControlTopic` configured. This avoids a failover window in which the old running job does not yet know the committed boundary. Once the job is running with boundary enforcement enabled, later forward-only plans may be applied while it runs. Run only one `apply` operation at a time; concurrent incomparable plans fail closed and require operator intervention. Do not recreate the action-state topic or change its partitions while an apply operation is running: the coordinator checks the topic ID and partition set before and after deletion, but Kafka addresses `deleteRecords` by topic name and cannot atomically fence topic lifecycle changes. - -The action-state topic and control topic must be dedicated to one job's recovery history. Once any boundary has been committed, every future run and restore of that history must retain the same control-topic configuration; omitting or changing it removes logical boundary enforcement. The boundary can move only forward. A checkpoint whose marker is below the committed boundary is rejected even if Kafka has not finished physical deletion. - -Per-key tombstones and checkpoint-aligned cleanup are mutually exclusive. Tombstones are not tied to the selected recovery boundary and could invalidate a checkpoint that checkpoint-aligned cleanup promises to retain. - -To migrate a job that previously emitted tombstones, first stop it cleanly so its Kafka producer closes and all accepted tombstone sends finish. Restart from the newest recovery point that is valid under the tombstone mode's existing recovery trade-off, with `kafkaActionStateTombstoneEnabled=false` and no cleanup control topic configured. After that attempt completes a new checkpoint or savepoint, stop it again, create and apply the first cleanup plan from that new recovery point, and restart from the same point with `kafkaActionStateCleanupControlTopic` configured. The new marker is after every old tombstone; selecting an older recovery point cannot provide the same guarantee because replay may still encounter those tombstones. +`KafkaActionStateCleanupTool` requires the following configuration and recovery Review Comment: Could we move the remaining “Checkpoint-aligned Kafka cleanup” subsection into the Consistency guide, including its prerequisites, topic policies, and CLI options? The cleanup procedure currently sends readers back to Configuration for these details. Keeping them together would make the procedure self-contained, while Configuration retains concise option descriptions and a link to the guide. -- 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]
