Async Request в YTsaurus Flow (Java)

Данный пример пайплайна реализует событийно-ориентированный цикл запрос–ответ. Входящие события порождают запросы к обработчику, ответы возвращаются обратно в тот же компьютейшен и накапливаются во внешнем стейте. Пример демонстрирует циклическую топологию пайплайна и совместное использование внешнего стейта.

Исходный код (Java)

Исходный код (Kotlin)

Компоненты

StateKeeperFunction

Обрабатывает два входных стрима — event и response. По событию event генерирует запрос с уникальным request_id и эмитирует его в стрим request. По событию response накапливает суммарную длину ответов в поле total_length внешнего стейта:

Java

Kotlin

    @Override
    public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
        String streamId = message.getStreamId();

        if ("event".equals(streamId)) {
            handleEvent(message, output, ctx);
        } else if ("response".equals(streamId)) {
            handleResponse(message, ctx);
        } else {
            throw new IllegalArgumentException("Unknown streamId: " + streamId);
        }
    }

    override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
        val streamId = message.getStreamId()

        if ("event" == streamId) {
            handleEvent(message, output, ctx)
        } else if ("response" == streamId) {
            handleResponse(message, ctx)
        } else {
            throw IllegalArgumentException("Unknown streamId: $streamId")
        }
    }

RequestProcessorFunction

Stateless-компьютейшен: получает запрос из стрима request, вычисляет длину строки и немедленно отправляет ответ в стрим response:

Java

Kotlin

    @Override
    public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
        long requestId = message.get("request_id", Long.class);
        long key = message.get("key", Long.class);
        String request = message.get("request", String.class);
        long length = request == null ? 0L : (long) request.length();

        log.debug("Processing request (requestId={}, key={}, length={})", requestId, key, length);

        var builder = ctx.createMessageBuilder("response");
        builder.set("request_id", requestId)
               .set("key", key)
               .set("length", length);

        output.addMessage(builder.finish());
    }

    override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
        val requestId = message.get("request_id", Long::class.java)
        val key = message.get("key", Long::class.java)
        val request = message.get("request", String::class.java)
        val length = if (request == null) 0L else request.length.toLong()

        log.debug("Processing request (requestId={}, key={}, length={})", requestId, key, length)

        val builder = ctx.createMessageBuilder("response")
        builder.set("request_id", requestId)
            .set("key", key)
            .set("length", length)

        output.addMessage(builder.finish())
    }

Регистрация компьютейшенов

Компьютейшены state и processor регистрируются аннотацией @FlowComputation на классах их process-функций:

Java

Kotlin

@FlowComputation(id = "state")

@FlowComputation(id = "processor")

@FlowComputation(id = "state")

@FlowComputation(id = "processor")

PipelineMain

Единственная точка входа (запускает пайплайн или обслуживает его как компаньон — по YT_FLOW_MODE):

Java

Kotlin

    public static void main(String[] args) throws Exception {
        new SpringApplicationBuilder(PipelineMain.class).run(args);
    }

        @JvmStatic
        fun main(args: Array<String>) {
            SpringApplicationBuilder(PipelineMain::class.java).run(*args)
        }

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

Циклическая топология — стрим response возвращается обратно в state-компьютейшен, замыкая цикл event → request → response → state. Flow поддерживает такие графы явно.
Маршрутизация по streamId — одна функция обрабатывает несколько входных стримов, определяя тип сообщения через message.getStreamId().
ExternalStateAccessor с PayloadBuilder — поле total_length обновляется точечно: current.toBuilder() → изменение → stateAccessor.set(updated.finish()).
Конфигурация через Spring Boot — компьютейшены регистрируются аннотацией @FlowComputation; flow-spring-boot-starter управляет жизненным циклом gRPC-сервера.

См. также

Shuffle

Retryable Async Request