---
metadata:
  - name: generator
    content: Diplodoc Platform v5.64.2
alternate:
  - https://ytsaurus.tech/docs/en/flow/concepts/failed-messages.md
  - https://ytsaurus.tech/docs/ru/flow/concepts/failed-messages.md
  - href: https://ytsaurus.tech/docs/en/flow/concepts/failed-messages.md
    type: text/markdown
    title: Markdown version
  - href: https://ytsaurus.tech/docs/en/llms.txt?revision=c426f39927f2807c73168512150ef54f4b741847
    rel: describedby
---
> **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/en/llms.txt?revision=c426f39927f2807c73168512150ef54f4b741847

<!-- source: en/_includes/flow/concepts/failed-messages.md -->
# Saving failed messages

If a message cannot be processed now and your application allows deferring its processing, save it with `output->AddFailedMessage(message, error)` and return from its handler. Flow writes it to the `<pipeline_path>/failed_messages` system table and continues processing other messages. This avoids repeated job failures caused by the same invalid data.

The mechanism is available for C++ functions running under the `NYT::NFlow::TProcessFunctionComputation` adapter inside the worker. Source, Swift, [process function graphs](https://ytsaurus.tech/docs/en/flow/cpp/process-functions.md#function-graph), and companions are not supported yet.

1. Enable failed-message capture at the root of the pipeline's [static spec](https://ytsaurus.tech/docs/en/flow/concepts/spec.md). It is disabled by default:

   ```yson
   {
       failure_handler = {
           enabled = %true;
       };
   }
   ```

2. In your process function, explicitly select the messages whose processing can be deferred. For example, for streams with an `int64` column named `value`, this handler saves messages with a negative value and forwards the rest to the `output` stream:

   ```cpp
   void ProcessMessage(
       const TInputMessageConstPtr& message,
       const IOutputCollectorPtr& output,
       const IRuntimeContextPtr& context) override
   {
       if (GetColumnValue<i64>(*message, "value") < 0) {
           output->AddFailedMessage(message, TError("value must be non-negative"));
           return;
       }

       output->AddMessage(context->ConvertToOutputMessage(*message, TStreamId("output")));
   }
   ```

Pass the original `message` received by `ProcessMessage`. Exceptions are not caught automatically: an unhandled exception still fails the epoch.

Writes to `failed_messages` commit atomically with the epoch's state and outputs. After a successful commit, the failed input is considered processed; if the write fails, the epoch does not commit. The table stores the payload, stream spec version ID (`stream_spec_id`), key, message and stream IDs, event timestamp, and the supplied error in `diagnostic`. The payload schema is stored in the pipeline's persistent stream spec history and is identified by `stream_spec_id`. The `failed_at` field contains the batch time in milliseconds.

`AddFailedMessage` does not undo earlier state changes, emitted messages, or timers. Validate the input before these actions if you need to defer them for a failed message.

Currently, the mechanism saves messages for later investigation. Repair and replay will be added separately; Flow cannot automatically return saved messages to processing yet.
<!-- endsource: en/_includes/flow/concepts/failed-messages.md -->