Retryable Async Request в YTsaurus Flow (C++)

Пайплайн аналогичен Async Request, но добавляет логику повторных попыток (ретраев) с использованием таймеров и внутреннего YsonState.

Исходный код

Отличие от Async Request

Ключевое отличие — TRequestProcessor запускается адаптером TProcessFunctionComputation (а не TProcessFunctionSwiftMapComputation), поскольку ему необходимо:

хранить внутренний стейт для отслеживания числа неудачных попыток;
использовать таймеры для повторных попыток через заданный интервал.

Компоненты пайплайна

TRequestProcessor

TRequestProcessor реализует IProcessFunction и использует TMutableStateKeyClient<TDelayedRequestState> для хранения внутреннего стейта (Internal YsonState). Стейт инициализируется в Init(const IRuntimeInitContextPtr& initContext) через initContext->InitClient(RequestStateClient_, "request_state").

При обработке входного сообщения (ProcessMessage):

Сохраняет запрос в стейт с FailedAttempts = 0
Вызывает TryRequest для выполнения попытки

Метод TryRequest содержит основную логику ретраев:

Если запрос "не удался" (определяется через IsRequestSuccessful), увеличивает счетчик FailedAttempts и ставит таймер через output->AddTimer(GetNextAttempt(context))
Если запрос "удался", создает TResponseMessage, сбрасывает стейт через state.Clear() и отправляет ответ

При срабатывании таймера (ProcessTimer) вызывается повторная попытка TryRequest с текущим стейтом.

TStateKeeper

Полностью аналогичен TStateKeeper из примера Async Request: принимает входные события и ответы, хранит аккумулированный результат во внешнем стейте.

Паттерн ретраев

Логика ретраев основана на следующих элементах:

TDelayedRequestState — наследник NYTree::TYsonStruct, хранит FailedAttempts и сам Request
TMutableStateKeyClient — клиент для работы с внутренним YsonState. В отличие от TSimpleExternalStateManager, стейт хранится во внутренних таблицах Flow, а не во внешней пользовательской таблице
Таймеры — при неудачной попытке устанавливается таймер с задержкой Delay через output->AddTimer(GetNextAttempt(context))
Константа MaxRetries = 3 задаёт период симуляции успешной попытки: запрос выполняется сразу или не более чем после двух ретраев. Delay = 5 задаёт задержку между попытками

Вычисление времени следующей попытки выполняется через context->GetCurrentTimestamp(), что обеспечивает корректную работу с системным временем Flow.

Структура пайплайна

events → TStateKeeper → request (генерация запроса)
request → TRequestProcessor → response (обработка с ретраями)
response → TStateKeeper → стейт (накопление результатов)

В спеке для TRequestProcessor необходимо зарегистрировать секцию timer_streams для поддержки повторных попыток. Класс process function указывается в поле processing_function.

Исходный код

TRequestProcessor

void TRequestProcessor::Init(const IRuntimeInitContextPtr& initContext)
{
    initContext->InitClient<TDelayedRequestState>(RequestStateClient_, "request_state");
}

void TRequestProcessor::ProcessMessage(
    const TInputMessageConstPtr& message,
    const IOutputCollectorPtr& output,
    const IRuntimeContextPtr& context)
{
    auto request = context->ConvertToYsonMessage<TRequestMessage>(message);
    auto state = RequestStateClient_.GetState(message->Key);
    state->Request = request;
    state->FailedAttempts = 0;
    TryRequest(state, output, context);
}

void TRequestProcessor::ProcessTimer(
    const TInputTimerConstPtr& timer,
    const IOutputCollectorPtr& output,
    const IRuntimeContextPtr& context)
{
    auto state = RequestStateClient_.GetState(timer->Key);
    TryRequest(state, output, context);
}

TSystemTimestamp TRequestProcessor::GetNextAttempt(const IRuntimeContextPtr& context) const
{
    return TSystemTimestamp{context->GetCurrentTimestamp().Underlying() + Delay};
}

bool TRequestProcessor::IsRequestSuccessful(ui64 requestId, i64 failedAttempts) const
{
    return (requestId + failedAttempts) % MaxRetries == 0;
}

void TRequestProcessor::TryRequest(
    TStateAccessor<TDelayedRequestState>& state,
    const IOutputCollectorPtr& output,
    const IRuntimeContextPtr& context) const
{
    if (!IsRequestSuccessful(state->Request->RequestId, state->FailedAttempts)) {
        state->FailedAttempts += 1;
        output->AddTimer(GetNextAttempt(context));
        YT_TLOG_DEBUG("Failed request")
            .With("RequestId", state->Request->RequestId)
            .With("FailedAttempts", state->FailedAttempts);
        return;
    }

    auto response = New<TResponseMessage>();
    response->RequestId = state->Request->RequestId;
    response->Key = state->Request->Key;
    response->Length = std::ssize(state->Request->Request);
    YT_TLOG_DEBUG("Processed request")
        .With("RequestId", state->Request->RequestId)
        .With("FailedAttempts", state->FailedAttempts);
    state.Clear();
    output->AddMessage(context->ConvertToMessage(response));
}

TStateKeeper

void TStateKeeper::Init(const IRuntimeInitContextPtr& initContext)
{
    initContext->InitExternalStateClient(StateClient_, "/state");
}

void TStateKeeper::ProcessMessage(
    const TInputMessageConstPtr& message,
    const IOutputCollectorPtr& output,
    const IRuntimeContextPtr& context)
{
    if (message->StreamId == "event") {
        auto event = context->ConvertToYsonMessage<TEventMessage>(message);
        auto request = New<TRequestMessage>();
        request->RequestId = RandomNumber<ui64>();
        request->Key = event->Key;
        request->Request = event->Data;
        output->AddMessage(context->ConvertToMessage(request));
        YT_TLOG_DEBUG("Send request")
            .With("RequestId", request->RequestId);
    } else if (message->StreamId == "response") {
        auto response = context->ConvertToYsonMessage<TResponseMessage>(message);
        YT_TLOG_DEBUG("Received response")
            .With("RequestId", response->RequestId);
        auto state = StateClient_.GetState(message->Key);
        i64 totalLength = state->GetColumnValue<std::optional<i64>>("total_length").value_or(0);
        totalLength += response->Length;
        TPayloadBuilder builder(state->Schema);
        builder.Set(totalLength, "total_length");
        state->Payload = builder.Finish();
    } else {
        THROW_ERROR_EXCEPTION("Unexpected stream_id")
            .With("stream_id", message->StreamId);
    }
}

См. также

Async Request

Wait Click Join