---
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/ru/flow/concepts/failed-messages.md
    type: text/markdown
    title: Markdown version
  - href: https://ytsaurus.tech/docs/ru/llms.txt?revision=c426f39927f2807c73168512150ef54f4b741847
    rel: describedby
---
> **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/ru/llms.txt?revision=c426f39927f2807c73168512150ef54f4b741847

<!-- source: ru/_includes/flow/concepts/failed-messages.md -->
# Сохранение проблемных сообщений

Если отдельное сообщение нельзя обработать сейчас, но по условиям вашей задачи его обработку допустимо отложить, сохраните его через `output->AddFailedMessage(message, error)` и завершите обработку этого сообщения. Flow запишет его в системную таблицу `<pipeline_path>/failed_messages` и продолжит обработку остальных сообщений. Это позволяет избежать повторных падений джобы на одних и тех же некорректных данных.

Механизм доступен для C++-функций, которые выполняются адаптером `NYT::NFlow::TProcessFunctionComputation` внутри воркера. Source, Swift, [графы process function](https://ytsaurus.tech/docs/ru/flow/cpp/process-functions.md#function-graph) и companions пока не поддерживаются.

1. В корне [статической спеки](https://ytsaurus.tech/docs/ru/flow/concepts/spec.md) пайплайна включите сохранение проблемных сообщений. По умолчанию оно выключено:

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

2. В process function явно выберите сообщения, обработку которых можно отложить. Например, для потоков с колонкой `value` типа `int64` следующий обработчик сохраняет сообщения с отрицательным значением, а остальные передаёт в выходной поток `output`:

   ```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")));
   }
   ```

Передавайте исходный `message`, полученный в `ProcessMessage`. Автоматического перехвата исключений нет: необработанное исключение по-прежнему приводит к ошибке эпохи.

Запись в `failed_messages` фиксируется атомарно с состоянием и выходами эпохи. После успешного коммита проблемное сообщение считается обработанным; при ошибке записи эпоха не фиксируется. В таблице сохраняются payload, идентификатор версии stream spec (`stream_spec_id`), ключ, идентификаторы сообщения и потока, event timestamp и переданная ошибка в `diagnostic`. Схема payload хранится в персистентной истории stream spec пайплайна и определяется по `stream_spec_id`. Поле `failed_at` содержит время батча в миллисекундах.

`AddFailedMessage` не отменяет уже сделанные изменения состояния, выходные сообщения и таймеры. Проверяйте данные до этих действий, если для проблемного сообщения их выполнение нужно отложить.

На текущем этапе реализовано сохранение сообщений для последующего разбора. Исправление и повторная обработка сохранённых сообщений будут добавлены отдельно; автоматически возвращать их в обработку Flow пока не умеет.
<!-- endsource: ru/_includes/flow/concepts/failed-messages.md -->