---
metadata:
  - name: generator
    content: Diplodoc Platform v5.63.3
alternate:
  - https://ytsaurus.tech/docs/en/flow/java/examples/async_request.md
  - https://ytsaurus.tech/docs/ru/flow/java/examples/async_request.md
  - href: https://ytsaurus.tech/docs/en/flow/java/examples/async_request.md
    type: text/markdown
    title: Markdown version
  - href: https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65
    rel: describedby
---
> **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65

<!-- source: en/_includes/flow/java/examples/async_request.md -->
# 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)](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/java/async_request)

[Source code (Kotlin)](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/kotlin/async_request)

## 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:

{% list tabs group=lang %}

- Java

  ```java
      @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);
          }
      }
  ```

- Kotlin

  ```kotlin
      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")
          }
      }
  ```

{% endlist %}

### 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:

{% list tabs group=lang %}

- Java

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

- Kotlin

  ```kotlin
      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())
      }
  ```

{% endlist %}

### Registering computations

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

{% list tabs group=lang %}

- Java

  ```java
  @FlowComputation(id = "state")
  ```

  ```java
  @FlowComputation(id = "processor")
  ```

- Kotlin

  ```kotlin
  @FlowComputation(id = "state")
  ```

  ```kotlin
  @FlowComputation(id = "processor")
  ```

{% endlist %}

### PipelineMain

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

{% list tabs group=lang %}

- Java

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

- Kotlin

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

{% endlist %}

## 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.
<!-- endsource: en/_includes/flow/java/examples/async_request.md -->

<!-- source: en/_includes/flow/java/examples/async_request_also.md -->
## See also

- [Quick start (Java)](https://ytsaurus.tech/docs/en/flow/java/getting-started.md)
- [Computation (Java)](https://ytsaurus.tech/docs/en/flow/java/computation.md)
- [Stateful processing](https://ytsaurus.tech/docs/en/flow/concepts/stateful.md)
- [Similar example in C++](https://ytsaurus.tech/docs/en/flow/cpp/examples/async_request.md)
- [Extended version with retries](https://ytsaurus.tech/docs/en/flow/java/examples/retryable_async_request.md)
<!-- endsource: en/_includes/flow/java/examples/async_request_also.md -->