Async Request in YTsaurus Flow (Java)

Use this pipeline example to implement an event-driven request–response cycle. Incoming events trigger requests to the handler. Responses go back to the same computation and accumulate in the external state. This example shows a cyclic pipeline topology and shared use of external state.

Source code (Java)

Source code (Kotlin)

Components

StateKeeperFunction

You handle two input streams — event and response. When you get an event, you generate a request with a unique request_id and emit it to the request stream. When you get a response, you accumulate the total length of responses in the total_length field of the external state:

    @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

This is a stateless computation: you get a request from the request stream, calculate the string length, and immediately send the response to the response stream:

    @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())
    }

Registering computations

You register the state and processor computations with the @FlowComputation annotation on their process function classes:

@FlowComputation(id = "state")

@FlowComputation(id = "processor")

@FlowComputation(id = "state")

@FlowComputation(id = "processor")

PipelineMain

This single entry point starts the pipeline or serves as its companion, depending on YT_FLOW_MODE.

    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)
        }

Key patterns

  • Cyclic topology: the response stream goes back to the state computation, closing the cycle event → request → response → state. Flow explicitly supports such graphs.
  • Routing by streamId: one function handles multiple input streams. You determine the message type using message.getStreamId().
  • ExternalStateAccessor with PayloadBuilder: you update the total_length field selectively: current.toBuilder() → make changes → stateAccessor.set(updated.finish()).
  • Configuration via Spring Boot: you register computations with the @FlowComputation annotation; flow-spring-boot-starter manages the gRPC server lifecycle.

See also

Previous
Shuffle
Next
Retryable Async Request