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

<!-- source: ru/_includes/flow/java/examples/wait_click_join.md -->
# Wait Click Join в YTsaurus Flow (Java)

[Пайплайн](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#pipeline) выполняет джойн [потоков](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#stream-and-computation) хитов (hit) и действий (action) на временном окне. Для каждого хита система ожидает связанные действия (show, click) в течение заданного временного интервала, после чего генерирует объединенное событие.

[Исходный код (Java)](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/java/wait_click_join)

[Исходный код (Kotlin)](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/kotlin/wait_click_join)
Подробное описание задачи и логики джойна — в [описании C++ версии](https://ytsaurus.tech/docs/ru/flow/cpp/examples/wait_click_join.md).

## Постановка задачи {#problem-statement}

Есть система с тремя видами событий:
- **hit** — факт запроса пользователя. Содержит `hit_id`, `hit_time` и `hit_payload`.
- **action show** — факт, что пользователь увидел ответ. Содержит `hit_id`, `hit_time`, `action_time` и `is_click = false`.
- **action click** — факт, что пользователь кликнул на ответ. Содержит `hit_id`, `hit_time`, `action_time` и `is_click = true`.

Данные поступают в двух потоках: `hit` и `action`.

На выходе строится джойн: к каждому show присоединяется информация о наличии клика и `hit_payload` с хита. Хиты без показов игнорируются. Система ждёт события action в течение `wait_for_actions` секунд после `hit_time`.

## Схема потоков данных {#data-flow}

```mermaid
flowchart TD
    hit_queue["Hit Queue<br/>(QYT)"]
    action_queue["Action Queue<br/>(QYT)"]
    joined_action_queue["Joined Action Queue<br/>(QYT)"]
    hit_reader["hit_reader<br/>(passthrough computation)"]
    action_reader["action_reader<br/>(passthrough computation)"]
    join_computation["join computation<br/>(with companion)"]
    state["external state<br/>(dynamic table)"]
    join_computation --> state
    state --> join_computation
    hit_queue --> hit_reader
    action_queue --> action_reader
    action_reader --> join_computation
    hit_reader --> join_computation
    join_computation --> joined_action_queue
```

## JoinProcessFunction

Основная логика пайплайна реализована в `JoinProcessFunction`, которая обрабатывает два потока — `hit` и `action`:

{% list tabs group=lang %}

- Java

  ```java
  @Override
  public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
      var streamId = message.getStreamId();
      String hitId;
      Long hitTime;
      Hit hit = null;
      Action action = null;
  
      if ("hit".equals(streamId)) {
          hit = message.getPayload();
          hitId = hit.getHitId();
          hitTime = hit.getHitTime();
      } else if ("action".equals(streamId)) {
          action = message.getPayload();
          hitId = action.getHitId();
          hitTime = action.getHitTime();
      } else {
          throw new IllegalArgumentException("Unknown streamId: %s".formatted(streamId));
      }
      log.debug("Process message (Key: {}, HitId: {}, HitTime: {})", message.getKey(), hitId, hitTime);
      long waitTime = parseDuration(ctx.getComputationParameters().get("wait_for_actions").stringValue()).toSeconds();
      // [BEGIN timer_setup]
      long maxTime = hitTime + waitTime;
      // All messages out of wait range would be dropped.
      if (message.getEventTimestamp() >= maxTime) {
          log.warn("Message is out of wait range, skip (HitId: {}, HitTime: {}, MaxTime: {})",
                  hitId, hitTime, maxTime);
          return;
      }
      // Drop late data.
      // [BEGIN late_data_check]
      if (message.getEventTimestamp() < ctx.getEpochInputEventWatermark()) {
          log.warn("Late message, skip (HitId: {}, EventTimestamp: {}, InputEventWatermark: {})",
                  hitId, message.getEventTimestamp(), ctx.getEpochInputEventWatermark());
          return;
      }
      // [END late_data_check]
      // [BEGIN state_update]
      ExternalStateAccessor stateAccessor = ctx.getState(JOINED_ACTION_STATE, message);
      PayloadBuilder joinState = stateAccessor.getOrDefault().toBuilder();
  
      if ("hit".equals(streamId)) {
          log.debug("Process hit message (HitId: {}, HitPayload: {}, Hit: {})", hitId, hit.getHitPayload(), hit);
          joinState.set("hit_payload", hit.getHitPayload());
      } else {
          log.debug("Process action message (HitId: {}, ActionTime: {}, Action: {})",
                  hitId, action.getActionTime(), action);
          if (Boolean.TRUE.equals(action.isClick())) {
              joinState.set("click_time", action.getActionTime());
          } else {
              joinState.set("show_time", action.getActionTime());
          }
      }
      log.debug("Set state (KeyHitId: {}, State: {})", hitId, joinState);
      stateAccessor.set(joinState.finish());
      // [END state_update]
      log.debug("Add timer for Key (MaxTime: {}, HitTime: {})", maxTime, hitTime);
      output.addTimer(maxTime, hitTime);
      // [END timer_setup]
  }
  // [END on_message]
  
  // [BEGIN on_timer]
  @Override
  public void onTimer(Timer timer, OutputCollector output, RuntimeContext ctx) {
      // Key to pojo conversion is not supported yet.
      String hitId = timer.getKey().get("hit_id", String.class);
      log.debug("Process timer (KeyHitId: {}, Timer: {})", hitId, timer);
      ExternalStateAccessor stateAccessor = ctx.getState(JOINED_ACTION_STATE, timer);
      Payload joinState = stateAccessor.get();
      if (joinState == null) {
          throw new IllegalStateException("Empty state for timer: " + timer);
      }
      log.debug("Process state (KeyHitId: {}, State: {})", hitId, joinState);
      if ("timer".equals(timer.getStreamId())) {
          // ExternalState to pojo conversion is not supported yet.
          if (joinState.get("show_time", Long.class) != null
                  && joinState.get("hit_payload", String.class) != null
                  && joinState.get("show_time", Long.class) != 0) {
              log.debug("Click join: {}", joinState);
              JoinedAction joinedAction = new JoinedAction();
              long hitTime = timer.getKey().get("hit_time", Long.class);
              joinedAction.setHitId(hitId);
              joinedAction.setHitTime(hitTime);
              joinedAction.setShowTime(joinState.get("show_time", Long.class));
              if (joinState.get("click_time", Long.class) != null && joinState.get("click_time", Long.class) != 0) {
                  joinedAction.setClick(true);
                  joinedAction.setClickTime(joinState.get("click_time", Long.class));
              } else {
                  joinedAction.setClick(false);
                  joinedAction.setClickTime(0L);
              }
              joinedAction.setHitPayload(joinState.get("hit_payload", String.class));
              output.addMessage(Message.builder().setStreamId("joined_action").setPayload(joinedAction).build());
          }
          stateAccessor.clear();
      } else {
          throw new IllegalArgumentException("Unknown streamId: %s".formatted(timer.getStreamId()));
      }
  }
  ```

- Kotlin

  ```kotlin
  override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
      val streamId = message.streamId
      val hitId: String
      val hitTime: Long
      var hit: Hit? = null
      var action: Action? = null
  
      if ("hit" == streamId) {
          hit = message.getPayload<Hit>()
          hitId = hit.hitId!!
          hitTime = hit.hitTime!!
      } else if ("action" == streamId) {
          action = message.getPayload<Action>()
          hitId = action.hitId!!
          hitTime = action.hitTime!!
      } else {
          throw IllegalArgumentException("Unknown streamId: $streamId")
      }
      log.debug("Process message (Key: {}, HitId: {}, HitTime: {})", message.key, hitId, hitTime)
      val waitTime = parseDuration(
          ctx.computationParameters.get("wait_for_actions")!!.stringValue()
      ).toSeconds()
      // [BEGIN timer_setup]
      val maxTime = hitTime + waitTime
      // All messages out of wait range would be dropped.
      if (message.eventTimestamp >= maxTime) {
          log.warn(
              "Message is out of wait range, skip (HitId: {}, HitTime: {}, MaxTime: {})",
              hitId, hitTime, maxTime
          )
          return
      }
      // Drop late data.
      // [BEGIN late_data_check]
      if (message.eventTimestamp < ctx.epochInputEventWatermark) {
          log.warn(
              "Late message, skip (HitId: {}, EventTimestamp: {}, InputEventWatermark: {})",
              hitId, message.eventTimestamp, ctx.epochInputEventWatermark
          )
          return
      }
      // [END late_data_check]
      // [BEGIN state_update]
      val stateAccessor = ctx.getState(JOINED_ACTION_STATE, message)
      val joinState = stateAccessor.orDefault.toBuilder()
  
      if ("hit" == streamId) {
          log.debug(
              "Process hit message (HitId: {}, HitPayload: {}, Hit: {})",
              hitId, hit!!.hitPayload, hit
          )
          joinState.set("hit_payload", hit.hitPayload)
      } else {
          log.debug(
              "Process action message (HitId: {}, ActionTime: {}, Action: {})",
              hitId, action!!.actionTime, action
          )
          if (action.isClick() == true) {
              joinState.set("click_time", action.actionTime)
          } else {
              joinState.set("show_time", action.actionTime)
          }
      }
      log.debug("Set state (KeyHitId: {}, State: {})", hitId, joinState)
      stateAccessor.set(joinState.finish())
      // [END state_update]
      log.debug("Add timer for Key (MaxTime: {}, HitTime: {})", maxTime, hitTime)
      output.addTimer(maxTime, hitTime)
      // [END timer_setup]
  }
  // [END on_message]
  
  // [BEGIN on_timer]
  override fun onTimer(timer: Timer, output: OutputCollector, ctx: RuntimeContext) {
      // Key to pojo conversion is not supported yet.
      val hitId = timer.key.get("hit_id", String::class.java)
      log.debug("Process timer (KeyHitId: {}, Timer: {})", hitId, timer)
      val stateAccessor = ctx.getState(JOINED_ACTION_STATE, timer)
      val joinState = stateAccessor.get()
          ?: throw IllegalStateException("Empty state for timer: $timer")
      log.debug("Process state (KeyHitId: {}, State: {})", hitId, joinState)
      if ("timer" == timer.streamId) {
          // ExternalState to pojo conversion is not supported yet.
          if (joinState.get("show_time", Long::class.java) != null &&
              joinState.get("hit_payload", String::class.java) != null &&
              joinState.get("show_time", Long::class.java) != 0L
          ) {
              log.debug("Click join: {}", joinState)
              val joinedAction = JoinedAction()
              val hitTime = timer.key.get("hit_time", Long::class.java)
              joinedAction.hitId = hitId
              joinedAction.hitTime = hitTime
              joinedAction.showTime = joinState.get("show_time", Long::class.java)
              if (joinState.get("click_time", Long::class.java) != null &&
                  joinState.get("click_time", Long::class.java) != 0L
              ) {
                  joinedAction.isClick = true
                  joinedAction.clickTime = joinState.get("click_time", Long::class.java)
              } else {
                  joinedAction.isClick = false
                  joinedAction.clickTime = 0L
              }
              joinedAction.hitPayload = joinState.get("hit_payload", String::class.java)
              output.addMessage(
                  Message.builder().setStreamId("joined_action").setPayload(joinedAction).build()
              )
          }
          stateAccessor.clear()
      } else {
          throw IllegalArgumentException("Unknown streamId: ${timer.streamId}")
      }
  }
  ```

{% endlist %}

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

### Фильтрация late data {#late-data-check}

{% list tabs group=lang %}

- Java

  ```java
  if (message.getEventTimestamp() < ctx.getEpochInputEventWatermark()) {
      log.warn("Late message, skip (HitId: {}, EventTimestamp: {}, InputEventWatermark: {})",
              hitId, message.getEventTimestamp(), ctx.getEpochInputEventWatermark());
      return;
  }
  ```

- Kotlin

  ```kotlin
  if (message.eventTimestamp < ctx.epochInputEventWatermark) {
      log.warn(
          "Late message, skip (HitId: {}, EventTimestamp: {}, InputEventWatermark: {})",
          hitId, message.eventTimestamp, ctx.epochInputEventWatermark
      )
      return
  }
  ```

{% endlist %}

Сообщения, пришедшие с event-временем меньше текущего [вотермарка](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#timestamps-and-watermarks), считаются опоздавшими и игнорируются. Подробнее про вотермарки — в разделе [Watermarks & Timers](https://ytsaurus.tech/docs/ru/flow/concepts/watermarks.md).

### [Таймеры](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#timer) для закрытия окна

{% list tabs group=lang %}

- Java

  ```java
  long maxTime = hitTime + waitTime;
  // All messages out of wait range would be dropped.
  if (message.getEventTimestamp() >= maxTime) {
      log.warn("Message is out of wait range, skip (HitId: {}, HitTime: {}, MaxTime: {})",
              hitId, hitTime, maxTime);
      return;
  }
  // Drop late data.
  // [BEGIN late_data_check]
  if (message.getEventTimestamp() < ctx.getEpochInputEventWatermark()) {
      log.warn("Late message, skip (HitId: {}, EventTimestamp: {}, InputEventWatermark: {})",
              hitId, message.getEventTimestamp(), ctx.getEpochInputEventWatermark());
      return;
  }
  // [END late_data_check]
  // [BEGIN state_update]
  ExternalStateAccessor stateAccessor = ctx.getState(JOINED_ACTION_STATE, message);
  PayloadBuilder joinState = stateAccessor.getOrDefault().toBuilder();
  
  if ("hit".equals(streamId)) {
      log.debug("Process hit message (HitId: {}, HitPayload: {}, Hit: {})", hitId, hit.getHitPayload(), hit);
      joinState.set("hit_payload", hit.getHitPayload());
  } else {
      log.debug("Process action message (HitId: {}, ActionTime: {}, Action: {})",
              hitId, action.getActionTime(), action);
      if (Boolean.TRUE.equals(action.isClick())) {
          joinState.set("click_time", action.getActionTime());
      } else {
          joinState.set("show_time", action.getActionTime());
      }
  }
  log.debug("Set state (KeyHitId: {}, State: {})", hitId, joinState);
  stateAccessor.set(joinState.finish());
  // [END state_update]
  log.debug("Add timer for Key (MaxTime: {}, HitTime: {})", maxTime, hitTime);
  output.addTimer(maxTime, hitTime);
  ```

- Kotlin

  ```kotlin
  val maxTime = hitTime + waitTime
  // All messages out of wait range would be dropped.
  if (message.eventTimestamp >= maxTime) {
      log.warn(
          "Message is out of wait range, skip (HitId: {}, HitTime: {}, MaxTime: {})",
          hitId, hitTime, maxTime
      )
      return
  }
  // Drop late data.
  // [BEGIN late_data_check]
  if (message.eventTimestamp < ctx.epochInputEventWatermark) {
      log.warn(
          "Late message, skip (HitId: {}, EventTimestamp: {}, InputEventWatermark: {})",
          hitId, message.eventTimestamp, ctx.epochInputEventWatermark
      )
      return
  }
  // [END late_data_check]
  // [BEGIN state_update]
  val stateAccessor = ctx.getState(JOINED_ACTION_STATE, message)
  val joinState = stateAccessor.orDefault.toBuilder()
  
  if ("hit" == streamId) {
      log.debug(
          "Process hit message (HitId: {}, HitPayload: {}, Hit: {})",
          hitId, hit!!.hitPayload, hit
      )
      joinState.set("hit_payload", hit.hitPayload)
  } else {
      log.debug(
          "Process action message (HitId: {}, ActionTime: {}, Action: {})",
          hitId, action!!.actionTime, action
      )
      if (action.isClick() == true) {
          joinState.set("click_time", action.actionTime)
      } else {
          joinState.set("show_time", action.actionTime)
      }
  }
  log.debug("Set state (KeyHitId: {}, State: {})", hitId, joinState)
  stateAccessor.set(joinState.finish())
  // [END state_update]
  log.debug("Add timer for Key (MaxTime: {}, HitTime: {})", maxTime, hitTime)
  output.addTimer(maxTime, hitTime)
  ```

{% endlist %}

При получении хита устанавливается таймер на `hit_time + wait_for_actions`. Таймер сработает, когда вотермарк достигнет `maxTime`, то есть когда система будет уверена, что все события с `event_time < maxTime` уже обработаны.

### ExternalStateAccessor с PayloadBuilder

[Стейт](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#state) хранится во внешней динамической таблице. `PayloadBuilder` позволяет обновлять отдельные поля стейта без полной пересериализации:

{% list tabs group=lang %}

- Java

  ```java
  ExternalStateAccessor stateAccessor = ctx.getState(JOINED_ACTION_STATE, message);
  PayloadBuilder joinState = stateAccessor.getOrDefault().toBuilder();
  
  if ("hit".equals(streamId)) {
      log.debug("Process hit message (HitId: {}, HitPayload: {}, Hit: {})", hitId, hit.getHitPayload(), hit);
      joinState.set("hit_payload", hit.getHitPayload());
  } else {
      log.debug("Process action message (HitId: {}, ActionTime: {}, Action: {})",
              hitId, action.getActionTime(), action);
      if (Boolean.TRUE.equals(action.isClick())) {
          joinState.set("click_time", action.getActionTime());
      } else {
          joinState.set("show_time", action.getActionTime());
      }
  }
  log.debug("Set state (KeyHitId: {}, State: {})", hitId, joinState);
  stateAccessor.set(joinState.finish());
  ```

- Kotlin

  ```kotlin
  val stateAccessor = ctx.getState(JOINED_ACTION_STATE, message)
  val joinState = stateAccessor.orDefault.toBuilder()
  
  if ("hit" == streamId) {
      log.debug(
          "Process hit message (HitId: {}, HitPayload: {}, Hit: {})",
          hitId, hit!!.hitPayload, hit
      )
      joinState.set("hit_payload", hit.hitPayload)
  } else {
      log.debug(
          "Process action message (HitId: {}, ActionTime: {}, Action: {})",
          hitId, action!!.actionTime, action
      )
      if (action.isClick() == true) {
          joinState.set("click_time", action.actionTime)
      } else {
          joinState.set("show_time", action.actionTime)
      }
  }
  log.debug("Set state (KeyHitId: {}, State: {})", hitId, joinState)
  stateAccessor.set(joinState.finish())
  ```

{% endlist %}

### Очистка стейта

После срабатывания таймера и генерации выходного события стейт удаляется через `stateAccessor.clear()`, чтобы не накапливать устаревшие профили.

## Модели данных {#models}

Для типизированной работы с сообщениями определены POJO-классы с JPA-аннотациями.

Схемы стримов этого примера выводятся из моделей: в `pipeline.yson` секции `streams` нет вовсе — runner подставляет схемы зарегистрированных типизированных стримов сам, как это делают C++ и Go SDK. Порядок полей модели задаёт порядок колонок, а `columnDefinition` фиксирует тип колонки там, где он отличается от типа по умолчанию для Java-типа: поле `String` по умолчанию выводится в `utf8`, поэтому колонки, которые в очередях объявлены как `string`, помечены `columnDefinition = "string"`.

Если схема стрима всё же задана в спеке, она имеет приоритет — runner ничего не перезаписывает.

<!-- source: ru/_includes/flow/java/_field_order_warning.md -->
{% note warning %}

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

Если стрим в спеке не описан, порядок колонок задаёт сама модель — тогда следить нужно за совместимостью с очередями и таблицами, в которые пишет пайплайн.

{% endnote %}
<!-- endsource: ru/_includes/flow/java/_field_order_warning.md -->

### Hit

{% list tabs group=lang %}

- Java

  ```java
  // The stream schema is derived from this class: the runner writes it into the pipeline spec,
  // so "hit" needs no schema of its own there. Field order defines column order, and
  // columnDefinition pins a column type where the default for the field type differs.
  @Entity
  public class Hit {
      @Column(name = "hit_id", columnDefinition = "string")
      private String hitId;
      @Column(name = "hit_time", columnDefinition = "uint64")
      private Long hitTime;
      @Column(name = "hit_payload", columnDefinition = "string")
      private String hitPayload;
  
      public Hit() {
      }
  
      public Hit(String hitId, Long hitTime, String hitPayload) {
          this.hitId = hitId;
          this.hitTime = hitTime;
          this.hitPayload = hitPayload;
      }
  
      public String getHitId() {
          return hitId;
      }
  
      public void setHitId(String hitId) {
          this.hitId = hitId;
      }
  
      public Long getHitTime() {
          return hitTime;
      }
  
      public void setHitTime(Long hitTime) {
          this.hitTime = hitTime;
      }
  
      public String getHitPayload() {
          return hitPayload;
      }
  
      public void setHitPayload(String hitPayload) {
          this.hitPayload = hitPayload;
      }
  
      @Override
      public String toString() {
          return "Hit{" +
                  "hitId='" + hitId + '\'' +
                  ", hitTime=" + hitTime +
                  ", hitPayload='" + hitPayload + '\'' +
                  '}';
      }
  }
  ```

- Kotlin

  ```kotlin
  // The stream schema is derived from this class: the runner writes it into the pipeline spec,
  // so "hit" needs no schema of its own there. Field order defines column order, and
  // columnDefinition pins a column type where the default for the field type differs.
  @Entity
  class Hit {
      @Column(name = "hit_id", columnDefinition = "string")
      var hitId: String? = null
  
      @Column(name = "hit_time", columnDefinition = "uint64")
      var hitTime: Long? = null
  
      @Column(name = "hit_payload", columnDefinition = "string")
      var hitPayload: String? = null
  
      constructor()
  
      constructor(hitId: String, hitTime: Long, hitPayload: String) {
          this.hitId = hitId
          this.hitTime = hitTime
          this.hitPayload = hitPayload
      }
  
      override fun toString(): String {
          return "Hit{" +
              "hitId='$hitId'" +
              ", hitTime=$hitTime" +
              ", hitPayload='$hitPayload'" +
              '}'
      }
  }
  ```

{% endlist %}

### Action

{% list tabs group=lang %}

- Java

  ```java
  // The stream schema is derived from this class: the runner writes it into the pipeline spec,
  // so "action" needs no schema of its own there. Field order defines column order, and
  // columnDefinition pins a column type where the default for the field type differs.
  @Entity
  public class Action {
      @Column(name = "hit_id", columnDefinition = "string")
      private String hitId;
      @Column(name = "hit_time", columnDefinition = "uint64")
      private Long hitTime;
      @Column(name = "is_click")
      private Boolean click;
      @Column(name = "action_time", columnDefinition = "uint64")
      private Long actionTime;
  
      public Action() {
      }
  
      public Action(String hitId, Long hitTime, Long actionTime, Boolean isClick) {
          this.hitId = hitId;
          this.hitTime = hitTime;
          this.actionTime = actionTime;
          this.click = isClick;
      }
  
      public String getHitId() {
          return hitId;
      }
  
      public void setHitId(String hitId) {
          this.hitId = hitId;
      }
  
      public Long getHitTime() {
          return hitTime;
      }
  
      public void setHitTime(Long hitTime) {
          this.hitTime = hitTime;
      }
  
      public Long getActionTime() {
          return actionTime;
      }
  
      public void setActionTime(Long actionTime) {
          this.actionTime = actionTime;
      }
  
      public Boolean isClick() {
          return click;
      }
  
      public void setIsClick(Boolean click) {
          this.click = click;
      }
  
      @Override
      public String toString() {
          return "Action{" +
                  "hitId='" + hitId + '\'' +
                  ", hitTime=" + hitTime +
                  ", actionTime=" + actionTime +
                  ", click=" + click +
                  '}';
      }
  }
  ```

- Kotlin

  ```kotlin
  // The stream schema is derived from this class: the runner writes it into the pipeline spec,
  // so "action" needs no schema of its own there. Field order defines column order, and
  // columnDefinition pins a column type where the default for the field type differs.
  @Entity
  class Action {
      @Column(name = "hit_id", columnDefinition = "string")
      var hitId: String? = null
  
      @Column(name = "hit_time", columnDefinition = "uint64")
      var hitTime: Long? = null
  
      @Column(name = "is_click")
      var click: Boolean? = null
  
      @Column(name = "action_time", columnDefinition = "uint64")
      var actionTime: Long? = null
  
      constructor()
  
      constructor(hitId: String, hitTime: Long, actionTime: Long, isClick: Boolean) {
          this.hitId = hitId
          this.hitTime = hitTime
          this.actionTime = actionTime
          this.click = isClick
      }
  
      fun isClick(): Boolean? = click
  
      fun setIsClick(click: Boolean?) {
          this.click = click
      }
  
      override fun toString(): String {
          return "Action{" +
              "hitId='$hitId'" +
              ", hitTime=$hitTime" +
              ", actionTime=$actionTime" +
              ", click=$click" +
              '}'
      }
  }
  ```

{% endlist %}

### JoinedAction

{% list tabs group=lang %}

- Java

  ```java
  // The stream schema is derived from this class: the runner writes it into the pipeline spec,
  // so "joined_action" needs no schema of its own there. Field order defines column order, and
  // columnDefinition pins a column type where the default for the field type differs.
  @Entity
  public class JoinedAction {
      @Column(name = "hit_id", columnDefinition = "string")
      private String hitId;
      @Column(name = "hit_time", columnDefinition = "uint64")
      private Long hitTime;
      @Column(name = "is_click")
      private Boolean isClick;
      @Column(name = "show_time", columnDefinition = "uint64")
      private Long showTime;
      @Column(name = "click_time", columnDefinition = "uint64")
      private Long clickTime;
      @Column(name = "hit_payload", columnDefinition = "string")
      private String hitPayload;
  
      public JoinedAction() {
      }
  
      public JoinedAction(String hitId, Long hitTime, Boolean isClick, Long showTime, Long clickTime, String hitPayload) {
          this.hitId = hitId;
          this.hitTime = hitTime;
          this.isClick = isClick;
          this.showTime = showTime;
          this.clickTime = clickTime;
          this.hitPayload = hitPayload;
      }
  
      public String getHitId() {
          return hitId;
      }
  
      public void setHitId(String hitId) {
          this.hitId = hitId;
      }
  
      public Long getHitTime() {
          return hitTime;
      }
  
      public void setHitTime(Long hitTime) {
          this.hitTime = hitTime;
      }
  
      public Boolean getClick() {
          return isClick;
      }
  
      public void setClick(Boolean click) {
          isClick = click;
      }
  
      public Long getShowTime() {
          return showTime;
      }
  
      public void setShowTime(Long showTime) {
          this.showTime = showTime;
      }
  
      public Long getClickTime() {
          return clickTime;
      }
  
      public void setClickTime(Long clickTime) {
          this.clickTime = clickTime;
      }
  
      public String getHitPayload() {
          return hitPayload;
      }
  
      public void setHitPayload(String hitPayload) {
          this.hitPayload = hitPayload;
      }
  }
  ```

- Kotlin

  ```kotlin
  // The stream schema is derived from this class: the runner writes it into the pipeline spec,
  // so "joined_action" needs no schema of its own there. Field order defines column order, and
  // columnDefinition pins a column type where the default for the field type differs.
  @Entity
  class JoinedAction {
      @Column(name = "hit_id", columnDefinition = "string")
      var hitId: String? = null
  
      @Column(name = "hit_time", columnDefinition = "uint64")
      var hitTime: Long? = null
  
      @get:JvmName("getClick")
      @set:JvmName("setClick")
      @Column(name = "is_click")
      var isClick: Boolean? = null
  
      @Column(name = "show_time", columnDefinition = "uint64")
      var showTime: Long? = null
  
      @Column(name = "click_time", columnDefinition = "uint64")
      var clickTime: Long? = null
  
      @Column(name = "hit_payload", columnDefinition = "string")
      var hitPayload: String? = null
  
      constructor()
  
      constructor(hitId: String, hitTime: Long, isClick: Boolean, showTime: Long, clickTime: Long, hitPayload: String) {
          this.hitId = hitId
          this.hitTime = hitTime
          this.isClick = isClick
          this.showTime = showTime
          this.clickTime = clickTime
          this.hitPayload = hitPayload
      }
  }
  ```

{% endlist %}

## Конфигурация пайплайна (Spring) {#pipeline-configuration}

[Исходный код (Java)](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/java/wait_click_join/wait_click_join/src/main/java/tech/ytsaurus/flow/examples/waitclickjoin/PipelineConfiguration.java)

[Исходный код (Kotlin)](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/kotlin/wait_click_join/wait_click_join/src/main/kotlin/tech/ytsaurus/flow/examples/waitclickjoin/PipelineConfiguration.kt)
Пример использует [Spring Boot интеграцию](https://ytsaurus.tech/docs/ru/flow/java/spring.md) для конфигурации. Компьютейшен `join` регистрируется аннотацией `@FlowComputation` на классе `JoinProcessFunction`:

{% list tabs group=lang %}

- Java

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

- Kotlin

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

{% endlist %}

Типизированные стримы объявляются через `ComputationProvider` (метод `getStreams()`):

{% list tabs group=lang %}

- Java

  ```java
  @Configuration
  public class PipelineConfiguration implements ComputationProvider {
  
      @Override
      public List<FlowStream<?>> getStreams() {
          return List.of(
                  FlowStreams.typed("hit", Hit.class),
                  FlowStreams.typed("action", Action.class),
                  FlowStreams.typed("joined_action", JoinedAction.class)
          );
      }
  }
  ```

- Kotlin

  ```kotlin
  @Configuration
  open class PipelineConfiguration : ComputationProvider {
  
      override fun getStreams(): List<FlowStream<*>> {
          return listOf(
              FlowStreams.typed("hit", Hit::class.java),
              FlowStreams.typed("action", Action::class.java),
              FlowStreams.typed("joined_action", JoinedAction::class.java)
          )
      }
  }
  ```

{% endlist %}

Ключевые моменты:
- Компьютейшен `join` (`JoinProcessFunction`) регистрируется аннотацией `@FlowComputation(id = "join")`.
- Регистрируются три типизированных стрима: `hit`, `action` и `joined_action`.
- Типизированные стримы позволяют использовать `message.getPayload()` для получения POJO-объектов.
- Этой регистрации достаточно, чтобы не описывать стримы в статической спеке: их схемы попадают в спеку при запуске.

## Точка входа {#entry-points}

### PipelineMain

Запускает пайплайн или обслуживает его как компаньон — роль выбирается по `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 %}

## Статическая спека {#static-spec}

### Computation join

```yson
"join" = {
    "computation_class_name" = "TJoin";
    "group_by_schema" = [
        {"name" = "hash"; "expression" = "farm_hash(hit_id)"; "type" = "uint64"};
        {"name" = "hit_id"; "type" = "string"};
        {"name" = "hit_time"; "type" = "uint64"};
    ];
    "input_stream_ids" = ["action"; "hit"];
    "output_stream_ids" = ["joined_action"];
    "external_state_managers" = {
        "/join-state" = {
            "external_state_manager_class_name" = "NYT::NFlow::TSimpleExternalStateManager";
            "parameters" = {
                "path" = "//path/to/state";
            };
        };
    };
    "parameters" = {
        "wait_for_actions" = "10s";
    };
    "timers" = {
        "timer" = {};
    };
};
```

- `group_by_schema` — ключ партиционирования: `(hash, hit_id, hit_time)`.
- `input_stream_ids` — два входных стрима: `action` и `hit`. Секции `streams` в спеке нет: схемы этих стримов и стрима `joined_action` выводятся из моделей данных (см. [Модели данных](#models)).
- `output_stream_ids` — один выходной стрим: `joined_action`.
- `external_state_managers` — top-level секция с описанием [External State](https://ytsaurus.tech/docs/ru/flow/java/external-state.md) (на одном уровне с `parameters`). Ключ (`"/join-state"`) — имя стейта, начинающееся с `/`; то же имя передаётся в дескриптор `StateDescriptors.external("/join-state")`. `external_state_manager_class_name` — зарегистрированный класс менеджера (`NYT::NFlow::TSimpleExternalStateManager` для стандартного варианта). `parameters/path` — путь к динамической таблице YTsaurus; схема ключевых колонок таблицы должна совпадать с `group_by_schema`.
- `parameters.wait_for_actions` — время ожидания событий action после хита. Доступен в Java через `RuntimeContext.getComputationParameters()`.
- `timers.timer` — объявление timer stream для закрытия хитов.

## Тестирование {#testing}

[Исходный код тестов (Java)](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/java/wait_click_join/wait_click_join/src/test/java/tech/ytsaurus/flow/examples/waitclickjoin/JoinProcessFunctionTest.java)

[Исходный код тестов (Kotlin)](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/kotlin/wait_click_join/wait_click_join/src/test/kotlin/tech/ytsaurus/flow/examples/waitclickjoin/JoinProcessFunctionTest.kt)
Для тестирования используется `TestComputationHarness` — тестовая обвязка, позволяющая вызывать `doProcess` без запуска полного пайплайна.

### Настройка тестового окружения

{% list tabs group=lang %}

- Java

  ```java
      @BeforeEach
      void init() throws IOException {
          var pipelineContext = new PipelineContext();
  
          Computation join = Computation.builder()
                  .setComputationId("join")
                  .setProcessFunction(new JoinProcessFunction())
                  .build();
          pipelineContext.registerComputation(join);
  
          pipelineContext.registerStream(FlowStreams.typed("hit", Hit.class));
          pipelineContext.registerStream(FlowStreams.typed("action", Action.class));
          pipelineContext.registerStream(FlowStreams.typed("joined_action", JoinedAction.class));
  
          // External state schema for "/join-state"
          this.joinStateSchema = TableSchema.builder()
                  .addValue("hit_payload", ColumnValueType.STRING)
                  .addValue("show_time", ColumnValueType.UINT64)
                  .addValue("click_time", ColumnValueType.UINT64)
                  .build();
          // The pipeline spec is packaged as a classpath resource in the app module's
          // src/main/resources, so the test reads the single source of truth directly.
          try (var specStream = getClass().getResourceAsStream("/pipeline.yson")) {
              YTreeNode spec = YTreeTextSerializer.deserialize(specStream);
              this.testHarness = TestComputationHarness.builder()
                      .setPipelineContext(pipelineContext)
                      .setPipelineSpec(spec)
                      .addExternalStateSchema("/join-state", joinStateSchema)
                      .build();
          }
      }
  ```

- Kotlin

  ```kotlin
      @BeforeEach
      fun init() {
          val pipelineContext = PipelineContext()
  
          val join = Computation.builder()
              .setComputationId("join")
              .setProcessFunction(JoinProcessFunction())
              .build()
          pipelineContext.registerComputation(join)
  
          pipelineContext.registerStream(FlowStreams.typed("hit", Hit::class.java))
          pipelineContext.registerStream(FlowStreams.typed("action", Action::class.java))
          pipelineContext.registerStream(FlowStreams.typed("joined_action", JoinedAction::class.java))
  
          // External state schema for "/join-state"
          this.joinStateSchema = TableSchema.builder()
              .addValue("hit_payload", ColumnValueType.STRING)
              .addValue("show_time", ColumnValueType.UINT64)
              .addValue("click_time", ColumnValueType.UINT64)
              .build()
          // The pipeline spec is packaged as a classpath resource in the app module's
          // src/main/resources, so the test reads the single source of truth directly.
          javaClass.getResourceAsStream("/pipeline.yson")!!.use { specStream ->
              val spec = YTreeTextSerializer.deserialize(specStream)
              this.testHarness = TestComputationHarness.builder()
                  .setPipelineContext(pipelineContext)
                  .setPipelineSpec(spec)
                  .addExternalStateSchema("/join-state", joinStateSchema)
                  .build()
          }
      }
  ```

{% endlist %}

### Тест обработки hit-сообщения

{% list tabs group=lang %}

- Java

  ```java
      @Test
      void testHitMessageStoresHitPayloadInState() {
          String hitId = "hit-1";
          long hitTime = BASE_HIT_TIME;
  
          var messages = List.of(buildHitMessage(hitId, hitTime, "payload-data"));
  
          var request = TestDoProcessRequest.builder("join")
                  .setMessages(messages)
                  .setWatermarks(defaultWatermarks())
                  .build();
  
          var response = testHarness.doProcess(request);
  
          // No output messages yet — timer is scheduled, output comes on timer fire
          assertTrue(response.getOutputMessagesFlatten().isEmpty());
  
          // A timer must be scheduled
          assertEquals(1, response.getOutputTimersFlatten().size());
          var timer = response.getOutputTimersFlatten().getFirst();
          assertEquals(hitTime + WAIT_SECONDS, timer.getTriggerTimestamp());
  
          // External state must be updated with hit_payload
          assertEquals(1, response.allStates().externalSize("/join-state"));
          Payload key = buildKey(hitId, hitTime);
          var state = response.allStates().get(JoinProcessFunction.JOINED_ACTION_STATE, key).get();
          assertEquals("payload-data", state.get("hit_payload", String.class));
      }
  ```

- Kotlin

  ```kotlin
      @Test
      fun testHitMessageStoresHitPayloadInState() {
          val hitId = "hit-1"
          val hitTime = BASE_HIT_TIME
  
          val messages = listOf(buildHitMessage(hitId, hitTime, "payload-data"))
  
          val request = TestDoProcessRequest.builder("join")
              .setMessages(messages)
              .setWatermarks(defaultWatermarks())
              .build()
  
          val response = testHarness.doProcess(request)
  
          // No output messages yet — timer is scheduled, output comes on timer fire
          assertTrue(response.outputMessagesFlatten.isEmpty())
  
          // A timer must be scheduled
          assertEquals(1, response.outputTimersFlatten.size)
          val timer = response.outputTimersFlatten.first()
          assertEquals(hitTime + WAIT_SECONDS, timer.triggerTimestamp)
  
          // External state must be updated with hit_payload
          assertEquals(1, response.allStates().externalSize("/join-state"))
          val key = buildKey(hitId, hitTime)
          val state = response.allStates().get(JoinProcessFunction.JOINED_ACTION_STATE, key).get()!!
          assertEquals("payload-data", state.get("hit_payload", String::class.java))
      }
  ```

{% endlist %}

### Тест полного цикла джойна

{% list tabs group=lang %}

- Java

  ```java
      @Test
      void testFullJoinFlow_HitThenShowThenClickThenTimer() {
          String hitId = "hit-20";
          long hitTime = BASE_HIT_TIME;
          long showTime = hitTime + 3L;
          long clickTime = hitTime + 7L;
  
          // Step 1: process hit message
          var hitMessages = List.of(buildHitMessage(hitId, hitTime, "full-payload"));
          var hitRequest = TestDoProcessRequest.builder("join")
                  .setMessages(hitMessages)
                  .setWatermarks(defaultWatermarks())
                  .build();
          var hitResponse = testHarness.doProcess(hitRequest);
  
          assertTrue(hitResponse.getOutputMessagesFlatten().isEmpty());
          assertEquals(1, hitResponse.getOutputTimersFlatten().size());
          assertEquals(1, hitResponse.allStates().externalSize("/join-state"));
  
          Payload key = buildKey(hitId, hitTime);
          Payload stateAfterHit = hitResponse.allStates()
                  .get(JoinProcessFunction.JOINED_ACTION_STATE, key).get();
          assertEquals("full-payload", stateAfterHit.get("hit_payload", String.class));
  
          // Step 2: process show action message, passing the state from step 1
          var showMessages = List.of(buildActionMessage(hitId, hitTime, showTime, false, hitTime + 1L));
          var showRequest = TestDoProcessRequest.builder("join")
                  .setMessages(showMessages)
                  .setState(JoinProcessFunction.JOINED_ACTION_STATE, key, stateAfterHit)
                  .setWatermarks(defaultWatermarks())
                  .build();
          var showResponse = testHarness.doProcess(showRequest);
  
          assertTrue(showResponse.getOutputMessagesFlatten().isEmpty());
          Payload stateAfterShow = showResponse.allStates().get(JoinProcessFunction.JOINED_ACTION_STATE, key)
                  .get();
          assertEquals("full-payload", stateAfterShow.get("hit_payload", String.class));
          assertEquals(showTime, stateAfterShow.get("show_time", Long.class));
  
          // Step 3: process click action message, passing the state from step 2
          var clickMessages = List.of(buildActionMessage(hitId, hitTime, clickTime, true, hitTime + 2L));
          var clickRequest = TestDoProcessRequest.builder("join")
                  .setMessages(clickMessages)
                  .setState(JoinProcessFunction.JOINED_ACTION_STATE, key, stateAfterShow)
                  .setWatermarks(defaultWatermarks())
                  .build();
          var clickResponse = testHarness.doProcess(clickRequest);
  
          assertTrue(clickResponse.getOutputMessagesFlatten().isEmpty());
          Payload stateAfterClick = clickResponse.allStates().get(JoinProcessFunction.JOINED_ACTION_STATE, key)
                  .get();
          assertEquals(clickTime, stateAfterClick.get("click_time", Long.class));
  
          // Step 4: fire the timer with the accumulated state
          Timer timer = buildTimer(hitId, hitTime);
          var timerRequest = TestDoProcessRequest.builder("join")
                  .setTimers(List.of(timer))
                  .setState(JoinProcessFunction.JOINED_ACTION_STATE, key, stateAfterClick)
                  .build();
          var timerResponse = testHarness.doProcess(timerRequest);
  
          assertEquals(1, timerResponse.getOutputMessagesFlatten().size());
          JoinedAction joinedAction = timerResponse.getOutputMessagesFlatten().getFirst().getPayload();
          assertEquals(hitId, joinedAction.getHitId());
          assertEquals(hitTime, joinedAction.getHitTime());
          assertEquals("full-payload", joinedAction.getHitPayload());
          assertEquals(showTime, joinedAction.getShowTime());
          assertTrue(joinedAction.getClick());
          assertEquals(clickTime, joinedAction.getClickTime());
  
          assertNull(timerResponse.allStates().get(JoinProcessFunction.JOINED_ACTION_STATE, key).get());
      }
  ```

- Kotlin

  ```kotlin
      @Test
      fun testFullJoinFlow_HitThenShowThenClickThenTimer() {
          val hitId = "hit-20"
          val hitTime = BASE_HIT_TIME
          val showTime = hitTime + 3L
          val clickTime = hitTime + 7L
  
          // Step 1: process hit message
          val hitMessages = listOf(buildHitMessage(hitId, hitTime, "full-payload"))
          val hitRequest = TestDoProcessRequest.builder("join")
              .setMessages(hitMessages)
              .setWatermarks(defaultWatermarks())
              .build()
          val hitResponse = testHarness.doProcess(hitRequest)
  
          assertTrue(hitResponse.outputMessagesFlatten.isEmpty())
          assertEquals(1, hitResponse.outputTimersFlatten.size)
          assertEquals(1, hitResponse.allStates().externalSize("/join-state"))
  
          val key = buildKey(hitId, hitTime)
          val stateAfterHit = hitResponse.allStates()
              .get(JoinProcessFunction.JOINED_ACTION_STATE, key).get()!!
          assertEquals("full-payload", stateAfterHit.get("hit_payload", String::class.java))
  
          // Step 2: process show action message, passing the state from step 1
          val showMessages = listOf(buildActionMessage(hitId, hitTime, showTime, false, hitTime + 1L))
          val showRequest = TestDoProcessRequest.builder("join")
              .setMessages(showMessages)
              .setState(JoinProcessFunction.JOINED_ACTION_STATE, key, stateAfterHit)
              .setWatermarks(defaultWatermarks())
              .build()
          val showResponse = testHarness.doProcess(showRequest)
  
          assertTrue(showResponse.outputMessagesFlatten.isEmpty())
          val stateAfterShow = showResponse.allStates()
              .get(JoinProcessFunction.JOINED_ACTION_STATE, key).get()!!
          assertEquals("full-payload", stateAfterShow.get("hit_payload", String::class.java))
          assertEquals(showTime, stateAfterShow.get("show_time", Long::class.java))
  
          // Step 3: process click action message, passing the state from step 2
          val clickMessages = listOf(buildActionMessage(hitId, hitTime, clickTime, true, hitTime + 2L))
          val clickRequest = TestDoProcessRequest.builder("join")
              .setMessages(clickMessages)
              .setState(JoinProcessFunction.JOINED_ACTION_STATE, key, stateAfterShow)
              .setWatermarks(defaultWatermarks())
              .build()
          val clickResponse = testHarness.doProcess(clickRequest)
  
          assertTrue(clickResponse.outputMessagesFlatten.isEmpty())
          val stateAfterClick = clickResponse.allStates()
              .get(JoinProcessFunction.JOINED_ACTION_STATE, key).get()!!
          assertEquals(clickTime, stateAfterClick.get("click_time", Long::class.java))
  
          // Step 4: fire the timer with the accumulated state
          val timer = buildTimer(hitId, hitTime)
          val timerRequest = TestDoProcessRequest.builder("join")
              .setTimers(listOf(timer))
              .setState(JoinProcessFunction.JOINED_ACTION_STATE, key, stateAfterClick)
              .build()
          val timerResponse = testHarness.doProcess(timerRequest)
  
          assertEquals(1, timerResponse.outputMessagesFlatten.size)
          val joinedAction = timerResponse.outputMessagesFlatten.first().getPayload<JoinedAction>()
          assertEquals(hitId, joinedAction.hitId)
          assertEquals(hitTime, joinedAction.hitTime)
          assertEquals("full-payload", joinedAction.hitPayload)
          assertEquals(showTime, joinedAction.showTime)
          assertTrue(joinedAction.isClick!!)
          assertEquals(clickTime, joinedAction.clickTime)
  
          assertNull(timerResponse.allStates().get(JoinProcessFunction.JOINED_ACTION_STATE, key).get())
      }
  ```

{% endlist %}

Полный набор тестов (9 сценариев) — в [исходном коде](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/java/wait_click_join/wait_click_join/src/test/java/tech/ytsaurus/flow/examples/waitclickjoin/JoinProcessFunctionTest.java).

## Интеграционное тестирование {#integration-testing}

Интеграционное тестирование Java-пайплайнов выполняется так же, как и для C++ пайплайнов — с запуском полного пайплайна, включая C++ воркеры, очереди и стримы. Подробнее см. [Интеграционные тесты](https://ytsaurus.tech/docs/ru/flow/java/testing.md).

## Ключевые идеи {#key-ideas}

1. **Временное окно**: Система ждёт события action в течение `wait_for_actions` секунд после `hit_time`. Все события вне этого окна отбрасываются.

2. **Watermark для late data**: События с `event_timestamp < watermark` считаются опоздавшими и отбрасываются. Это неизбежно приводит к потере некоторых данных, но обеспечивает корректность watermark.

3. **External State для накопления данных**: Промежуточные данные (hit_payload, show_time, click_time) хранятся в External State, привязанном к ключу `(hit_id, hit_time)`. В данном случае нет ограничений на использование только External State, реализация с Internal State также возможна.

4. **Таймер для закрытия окна**: Таймер с `triggerTimestamp = hit_time + wait_for_actions` обеспечивает генерацию результата после закрытия окна ожидания.

5. **Идемпотентность таймеров**: Таймер устанавливается при каждом сообщении, но с одним и тем же `triggerTimestamp`. Flow дедуплицирует таймеры с одинаковым ключом и `triggerTimestamp`.

6. **Очистка стейта**: После генерации результата стейт очищается через `stateAccessor.clear()`, что удаляет строку из таблицы.
<!-- endsource: ru/_includes/flow/java/examples/wait_click_join.md -->

<!-- source: ru/_includes/flow/java/examples/wait_click_join_also.md -->
## См. также

- [Быстрый старт (Java)](https://ytsaurus.tech/docs/ru/flow/java/getting-started.md)
- [Таймеры](https://ytsaurus.tech/docs/ru/flow/concepts/timers.md)
- [Stateful processing](https://ytsaurus.tech/docs/ru/flow/concepts/stateful.md)
<!-- endsource: ru/_includes/flow/java/examples/wait_click_join_also.md -->