Retryable Async Request в YTsaurus Flow (Python)

Расширение примера AsyncRequest: обработчик запросов поддерживает автоматические повторные попытки через таймеры. При неудаче запрос сохраняется во внутреннем стейте и повторяется через фиксированный интервал, вплоть до MAX_RETRIES попыток.

Исходный код

Структура

Пайплайн состоит из двух компьютейшенов:

state (StateKeeperFunction) — stateful-компьютейшен: маршрутизирует события в запросы и накапливает суммарную длину ответов во внешнем стейте (аналогично async_request).
processor (RequestProcessorFunction) — stateful-компьютейшен с логикой повторов:
- При первой попытке сохраняет запрос во внутреннем стейте и пытается его обработать.
- Если обработка не удалась — увеличивает счётчик неудач, сохраняет стейт и устанавливает таймер (DELAY_SECONDS = 5).
- По таймеру повторяет попытку (до MAX_RETRIES = 3).
- При успехе — отправляет ответ и очищает стейт.

`request_processor_function.py`

Ключевой файл примера: реализует логику повторных попыток через on_timer и вспомогательный метод _try_or_retry.

class RequestProcessorFunction(RowFunction):
    """Retries requests up to MAX_RETRIES times; emits response on success."""

    def on_message(self, message, output, ctx):
        request_id = message.payload["request_id"]
        data = {
            "request_id": request_id,
            "key": message.payload["key"],
            "request": message.payload["request"],
            "failed_attempts": 0,
        }
        state = ctx.state("request-state", message)
        self._try_or_retry(state, data, output, ctx)

    def on_timer(self, timer, output, ctx):
        state = ctx.state("request-state", timer)
        data = state.get_or_default(None)
        if data is not None:
            self._try_or_retry(state, data, output, ctx)

    def _try_or_retry(self, state, data, output, ctx):
        request_id = data["request_id"]
        failed_attempts = data["failed_attempts"]

        if not _is_request_successful(request_id, failed_attempts):
            data["failed_attempts"] += 1
            state.set(data)
            output.add_timer(int(time.time()) + DELAY_SECONDS)
            log.debug(
                "Failed request (RequestId: %s, FailedAttempts: %s)",
                request_id,
                data["failed_attempts"],
            )
        else:
            key = data["key"]
            length = len(data["request"])

            builder = ctx.message_builder("response")
            builder.set("request_id", request_id)
            builder.set("key", key)
            builder.set("length", length)
            output.add_message(builder.finish())

            state.clear()
            log.debug(
                "Processed request (RequestId: %s, FailedAttempts: %s)",
                request_id,
                failed_attempts,
            )

`state_keeper_function.py`

Маршрутизация стримов и накопление результатов во внешнем стейте (идентично async_request).

class StateKeeperFunction(RowFunction):
    """Routes event messages to requests and accumulates response lengths in external state."""

    def on_message(self, message, output, ctx):
        stream_id = message.stream_id

        if stream_id == "event":
            key = message.payload["key"]
            data = message.payload["data"]

            request_id = random.getrandbits(64)

            builder = ctx.message_builder("request")
            builder.set("request_id", request_id)
            builder.set("key", key)
            builder.set("request", data)
            output.add_message(builder.finish())

            log.debug("Send request (RequestId: %s)", request_id)

        elif stream_id == "response":
            request_id = message.payload["request_id"]
            length = message.payload["length"]

            log.debug("Received response (RequestId: %s)", request_id)

            state = ctx.external_state("/state", message)
            current = state.get("total_length")
            total_length = current if current is not None else 0
            total_length += length

            builder = state.to_builder()
            builder.set("total_length", total_length)
            state.set(builder.finish())

        else:
            raise ValueError(f"Unexpected stream_id: {stream_id}")

`main.py`

Точка входа: структура идентична async_request, отличается только импортируемыми классами.

def main():
    log.info("Starting companion execution")

    pipeline = Pipeline()
    pipeline.add("state", StateKeeperFunction())
    pipeline.add("processor", RequestProcessorFunction())

    log.info("Starting pipeline...")
    pipeline.run()
    log.info("Pipeline completed")

Ключевые паттерны

Retry через таймеры: output.add_timer(int(time.time()) + DELAY_SECONDS) откладывает повторную попытку; on_timer читает стейт и повторяет _try_or_retry — стандартный паттерн реализации повторов в Flow без внешних очередей.
Счётчик попыток в стейте: поле failed_attempts сохраняется вместе с данными запроса в ctx.state("request-state", message), что гарантирует корректность при рестарте.
Детерминированная симуляция сбоев: _is_request_succeed(request_id, failed_attempts) моделирует нестабильный внешний сервис; в реальном коде заменяется вызовом HTTP-клиента.
Очистка стейта при успехе: state.clear() после успешной обработки предотвращает повторное срабатывание таймера на устаревших данных.
Разделение ответственности: StateKeeperFunction не знает о повторах — вся логика retry инкапсулирована в RequestProcessorFunction, что упрощает замену стратегии повторов.

См. также

Async Request

Wait Click Join