---
metadata:
  - name: generator
    content: Diplodoc Platform v5.63.3
alternate:
  - https://ytsaurus.tech/docs/en/flow/java/examples/url_downloader.md
  - https://ytsaurus.tech/docs/ru/flow/java/examples/url_downloader.md
  - href: https://ytsaurus.tech/docs/en/flow/java/examples/url_downloader.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/url_downloader.md -->
# URL Downloader in YTsaurus Flow (Java)

The [pipeline](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#pipeline) groups incoming URLs by host, accumulates them in the internal [state](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#state), and processes them in batches based on a timer. The timer triggers 5 seconds after a URL is added to the queue and emits the results to the output [stream](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#stream-and-computation).

[Source code (Java)](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/java/url_downloader)

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

## Components

### UrlDownloadFunction

This is the main process function that implements the logic for accumulating and processing URLs. The `onMessage` method adds a URL to the host’s internal state and sets a timer. The `onTimer` method processes the accumulated URLs and emits the results.

{% list tabs group=lang %}

- Java

  ```java
      @Override
      public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
          UrlMessage urlMessage = message.getPayload();
          String host = urlMessage.getHost();
          String url = urlMessage.getUrl();
  
          StateAccessor<HostState> accessor = ctx.getState(HOST_STATE, message);
          HostState state = accessor.getOrDefault(new HostState(host));
          state.getPendingUrls().add(url);
  
          output.addTimer(System.currentTimeMillis() / 1000 + 5, 0L);
          log.debug("Queued url (Host: {}, Url: {})", host, url);
      }
  ```

- Kotlin

  ```kotlin
      override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
          val urlMessage: UrlMessage = message.getPayload()
          val host = urlMessage.host
          val url = urlMessage.url
  
          val accessor = ctx.getState(HOST_STATE, message)
          val state = accessor.getOrDefault(HostState(host))
          state.pendingUrls!!.add(url)
  
          output.addTimer(System.currentTimeMillis() / 1000 + 5, 0L)
          log.debug("Queued url (Host: {}, Url: {})", host, url)
      }
  ```

{% endlist %}

{% list tabs group=lang %}

- Java

  ```java
      @Override
      public void onTimer(Timer timer, OutputCollector output, RuntimeContext ctx) {
          StateAccessor<HostState> accessor = ctx.getState(HOST_STATE, timer);
          HostState state = accessor.get();
          if (state == null || state.getPendingUrls().isEmpty()) {
              accessor.clear();
              return;
          }
          String host = state.getHost();
  
          for (String url : new ArrayList<>(state.getPendingUrls())) {
              String data = processUrl(url);
              ProcessedUrl processedUrl = new ProcessedUrl(host, url, data);
              output.addMessage(new Message("processed_urls", processedUrl));
              log.debug("Processed url (Host: {}, Url: {})", host, url);
          }
          accessor.clear();
      }
  ```

- Kotlin

  ```kotlin
      override fun onTimer(timer: Timer, output: OutputCollector, ctx: RuntimeContext) {
          val accessor = ctx.getState(HOST_STATE, timer)
          val state = accessor.get()
          if (state == null || state.pendingUrls.isNullOrEmpty()) {
              accessor.clear()
              return
          }
          val host = state.host
  
          for (url in ArrayList(state.pendingUrls)) {
              val data = processUrl(url!!)
              val processedUrl = ProcessedUrl(host, url, data)
              output.addMessage(Message("processed_urls", processedUrl))
              log.debug("Processed url (Host: {}, Url: {})", host, url)
          }
          accessor.clear()
      }
  ```

{% endlist %}

### HostState

This is the internal state model that is serialized to YSON. It stores the host name and the list of URLs that are waiting to be processed.

{% list tabs group=lang %}

- Java

  ```java
  @Entity
  public class HostState {
      private String host;
  
      @Column(name = "pending_urls")
      private List<String> pendingUrls = new ArrayList<>();
  
      public HostState() {
      }
  
      public HostState(String host) {
          this.host = host;
      }
  
      public String getHost() {
          return host;
      }
  
      public void setHost(String host) {
          this.host = host;
      }
  
      public List<String> getPendingUrls() {
          return pendingUrls;
      }
  
      public void setPendingUrls(List<String> urls) {
          this.pendingUrls = urls;
      }
  }
  ```

- Kotlin

  ```kotlin
  @Entity
  class HostState {
      var host: String? = null
  
      @Column(name = "pending_urls")
      var pendingUrls: MutableList<String?>? = ArrayList()
  
      constructor()
  
      constructor(host: String?) {
          this.host = host
      }
  }
  ```

{% endlist %}

### Registering the computation and streams

You register the `url_downloader` computation with the `@FlowComputation` annotation on the process function class.

{% list tabs group=lang %}

- Java

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

- Kotlin

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

{% endlist %}

You declare typed streams via `ComputationProvider` (the `getStreams()` method).

{% list tabs group=lang %}

- Java

  ```java
  @Configuration
  public class UrlDownloaderComputationContext implements ComputationProvider {
  
      @Override
      public List<FlowStream<?>> getStreams() {
          return List.of(
                  FlowStreams.typed("urls", UrlMessage.class),
                  FlowStreams.typed("processed_urls", ProcessedUrl.class)
          );
      }
  }
  ```

- Kotlin

  ```kotlin
  @Configuration
  open class UrlDownloaderComputationContext : ComputationProvider {
  
      override fun getStreams(): List<FlowStream<*>> {
          return listOf(
              FlowStreams.typed("urls", UrlMessage::class.java),
              FlowStreams.typed("processed_urls", ProcessedUrl::class.java),
          )
      }
  }
  ```

{% 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

- **Grouping by key**: you create a separate state for each host. Flow automatically routes messages with the same key to a single computation instance.
- **Wall-clock time-based timer**: `output.addTimer(System.currentTimeMillis() / 1000 + 5, 0L)` starts processing 5 seconds after a URL is added to the queue. Multiple `addTimer` calls with the same `triggerTimestamp` are deduplicated.
- **Batch processing in `onTimer`**: all accumulated URLs are processed at once when the timer triggers. This reduces the number of calls to downstream services.
- **State cleanup**: after processing, you delete the state via `accessor.clear()` to prevent memory leaks.
- **YsonStateAccessor**: the internal state is serialized to YSON and stored on the C++ worker side. You get the Java object via `getOrDefault`.
<!-- endsource: en/_includes/flow/java/examples/url_downloader.md -->

<!-- source: en/_includes/flow/java/examples/url_downloader_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/url_downloader.md)
<!-- endsource: en/_includes/flow/java/examples/url_downloader_also.md -->