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

<!-- source: en/_includes/flow/java/examples/external_state_join.md -->
# External State Join in YTsaurus Flow (Java and Kotlin)

This pipeline enriches queue events from a prebuilt dynamic reference table. Java and Kotlin use the same read-only joined-state API.

[Java source](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/java/external_state_join)

[Kotlin source](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/kotlin/external_state_join)

## Pipeline structure {#structure}

`event_reader` publishes the `event` stream. The `lookup_join` companion computation is grouped by `key`, reads `/reference` through `external_state_joiners`, and sends successful matches to `enriched`. The queue sink stores that output. The joiner's `path` may point to a Cypress link, allowing the reference table to be switched without restarting the pipeline.

See the [Java pipeline spec](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/java/external_state_join/external_state_join/src/main/resources/pipeline.yson) or [Kotlin pipeline spec](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/kotlin/external_state_join/external_state_join/src/main/resources/pipeline.yson).

## Read the joined state {#lookup}

`StateDescriptors.externalReadOnly("/reference")` binds the process function to the joiner declared under the same name. The accessor selects the row by the current message key. A missing row returns no `name`, so the example emits nothing. The joined accessor does not provide a write path.

{% list tabs group=lang %}

- Java

  ```java
  @Override
  public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
      ReadOnlyExternalStateAccessor stateAccessor = ctx.getState(REFERENCE_STATE, message);
      String name = stateAccessor.getOrDefault().get("name", String.class);
      if (name == null) {
          return;
      }
  
      output.addMessage(ctx.createMessageBuilder("enriched")
              .set("key", message.get("key", Long.class))
              .set("name", name)
              .finish());
  }
  ```

- Kotlin

  ```kotlin
  override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
      val stateAccessor = ctx.getState(REFERENCE_STATE, message)
      val name = stateAccessor.orDefault.get("name", String::class.java) ?: return
  
      output.addMessage(
          ctx.createMessageBuilder("enriched")
              .set("key", message.get("key", Long::class.java))
              .set("name", name)
              .finish()
      )
  }
  ```

{% endlist %}

The function is registered as `lookup_join` through `@FlowComputation`; the pipeline spec declares the `event` and `enriched` stream schemas.

## See also

- [External State (Java)](https://ytsaurus.tech/docs/en/flow/java/external-state.md)
- [Static Table Join example](https://ytsaurus.tech/docs/en/flow/java/examples/static_table_join.md)
<!-- endsource: en/_includes/flow/java/examples/external_state_join.md -->