---
metadata:
  - name: generator
    content: Diplodoc Platform v5.64.2
alternate:
  - https://ytsaurus.tech/docs/en/flow/java/examples/static_table_join.md
  - https://ytsaurus.tech/docs/ru/flow/java/examples/static_table_join.md
  - href: https://ytsaurus.tech/docs/en/flow/java/examples/static_table_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/static_table_join.md -->
# Static Table Join in YTsaurus Flow (Java and Kotlin)

This pipeline loads a static reference table into external state and enriches queue events by the shared key. Java and Kotlin use the same SDK and pipeline topology.

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

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

## Pipeline structure {#structure}

- `reference_reader` reads the static reference table;
- `reference_loader` normalizes each reference value and writes it to `/reference_state` through `external_state_managers`;
- `event_reader` reads events from a queue;
- `enricher` reads `/reference_state` through `external_state_joiners` and publishes only successful matches to `enriched`.

The manager and joiner point to the same dynamic state table. Their keys must match the `group_by_schema` of their computations. See the [Java pipeline spec](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/java/static_table_join/static_table_join/src/main/resources/pipeline.yson) or [Kotlin pipeline spec](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/kotlin/static_table_join/static_table_join/src/main/resources/pipeline.yson).

## Load the reference state {#load}

{% list tabs group=lang %}

- Java

  ```java
  @Override
  public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
      String name = message.get("name", String.class);
      String normalizedName = name.strip().toLowerCase();
  
      ExternalStateAccessor stateAccessor = ctx.getState(REFERENCE_STATE, message);
      stateAccessor.set(stateAccessor.getOrDefault().toBuilder()
              .set("normalized_name", normalizedName)
              .finish());
  }
  ```

- Kotlin

  ```kotlin
  override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
      val normalizedName = message.get("name", String::class.java)?.trim()?.lowercase() ?: return
  
      val stateAccessor = ctx.getState(REFERENCE_STATE, message)
      stateAccessor.set(
          stateAccessor.orDefault.toBuilder()
              .set("normalized_name", normalizedName)
              .finish()
      )
  }
  ```

{% endlist %}

## Enrich events {#enrich}

The read-only descriptor selects the joined row by the current message key. If the row or `normalized_name` is absent, the example produces no output.

{% list tabs group=lang %}

- Java

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

- Kotlin

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

{% endlist %}

Both computations are discovered through `@FlowComputation`; the pipeline spec declares the three stream schemas used by the native sources and companion functions.

## See also

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