Сохранение проблемных сообщений

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

Механизм доступен для C++-функций, которые выполняются адаптером NYT::NFlow::TProcessFunctionComputation внутри воркера. Source, Swift, графы process function и companions пока не поддерживаются.

В корне статической спеки пайплайна включите сохранение проблемных сообщений. По умолчанию оно выключено:
```
{
    failure_handler = {
        enabled = %true;
    };
}
```

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

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 пока не умеет.

Гарантии обработки

Порядок обработки сообщений