---
metadata:
  - name: generator
    content: Diplodoc Platform v5.63.3
alternate:
  - https://ytsaurus.tech/docs/en/flow/connectors/queue.md
  - https://ytsaurus.tech/docs/ru/flow/connectors/queue.md
  - href: https://ytsaurus.tech/docs/en/flow/connectors/queue.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/connectors/queue.md -->
# QYT Connector in YTsaurus Flow

This connector works with YTsaurus [queues](https://ytsaurus.tech/docs/en/user-guide/dynamic-tables/queues.md) (QYT).

You can find the connector code [here](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/connectors/queue).

## Reading from a Queue

When you read from a queue using the simplest method, messages arrive in [`Computation`](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#stream-and-computation) with the same schema as the queue table rows.

By default, the message’s `event_time` and `system_time` are calculated based on the `$timestamp` column, which contains the YT timestamp of the row record (for more details, see the [documentation for ordered dynamic tables](https://ytsaurus.tech/docs/en/user-guide/dynamic-tables/queues.md)).
But if the queue has a column with metadata (see the static spec parameters for the source), the message’s `event_time` can be taken from the corresponding field. You can also get stream watermark information from the metadata.

The [source](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#source) class is `NYT::NFlow::TQueueSource`.

Below are the source settings.

{% note info %}

The source is defined by the cluster and the queue path (`queue_path`). If you change them, the partitions are recreated and read again — see [Changing the Source](https://ytsaurus.tech/docs/en/flow/connectors/about.md#source-change).

{% endnote %}

### Managing Consumer Lag

When you register a consumer for a queue, the consumer lag is set to the last stored data in the queue. So, after registration, if the policy for deleting old rows in the queue is 1 day, the lag will also be 1 day.

This behavior can lead to the following scenario: when you roll out a release, the data lag will be 1 day, and message processing will pause until the lag is processed.

You can avoid this in several ways:

#### 1. Using the Standard YT CLI

Use the standard YT CLI to set the consumer offsets to the desired values for each partition. This isn’t very convenient for the first read of a topic, because you need to know the value of the latest offset and set it for each [partition](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#partition).

Here’s an example command:
```bash
yt advance-queue-consumer --proxy <cluster-name> //home/service/stable/consumer <cluster-name>://home/source_service/Data/queue --partition-index 0 --new-offset 486482013
```

For more advanced options, check the YT documentation or use the `-h` parameter.


### Static spec

<!-- source: en/flow/generated_docs/NYT_NFlow_TUnitedParameters_NYT_NFlow_TQueueSource.md -->
<!-- This file is generated by yt/yt/flow/yandex/tools/generate_yson_struct_doc/generate.sh script -->
<!-- Before using doc generation tool check readme: yt/yt/flow/yandex/tools/generate_yson_struct_doc/README.md -->
Source: [yt/yt/flow/library/cpp/common/registry-inl.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/common/registry-inl.h)

#|
|| **Parameter** | **Description** ||
|| `finite` | **Type**: `bool`
**Default value**: `false`
Treat the source as finite: transition the stream to the completed state after reading all available messages. How the end of the data is determined depends on the connector. ||
|| `queue_path` | **Type**: `NYT::NYPath::TRichYPath`
**Required parameter**
Path to the queue including the cluster. ||
|| `consumer_path` | **Type**: `NYT::NYPath::TRichYPath`
**Required parameter**
Path to the queue consumer with the cluster specified. ||
|| `try_parse_flow_queue_meta` | **Type**: `bool`
**Default value**: `true`
Whether to retrieve and parse meta from the queue (the meta contains watermarks provided by the writer to the queue). ||
|| `flow_queue_meta_column` | **Type**: `std::string`
**Default value**: `flow_queue_meta`
Column from which to retrieve meta. ||
|| `ignore_malformed_flow_queue_meta` | **Type**: `bool`
**Default value**: `false`
Whether to ignore invalid meta or fail. ||
|| `partition_filter` | **Type**: `std::optional<std::vector<std::pair<int, int>>>`
 ||
|#


{% cut "**Additional parameters**" %}


#|
|| `update_info_period` | **Type**: [TDuration](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#TDuration)
**Default value**: `15s`
Period for refreshing the source's auxiliary partition information. A connector may use this tick for status requests and read-session liveness checks. ||
|| `byte_size_alpha` | **Type**: `double`
**Default value**: `0.05`
Exponential smoothing coefficient for the average byte total and message count per offset: larger values make the estimate react to new data faster. ||
|| `update_partition_count_period` | **Type**: [TDuration](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#TDuration)
**Default value**: `1m`
How often the controller should update the partition count in the queue (the controller changes the computation partition count according to the queue partition count). ||
|| `update_partition_count_retry_min_backoff` | **Type**: [TDuration](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#TDuration)
**Default value**: `1s`
Initial delay before retrying a failed partition count update, subject to jitter. The value is capped by `update_partition_count_period`; subsequent delays increase exponentially. ||
|#


{% endcut %}
<!-- endsource: en/flow/generated_docs/NYT_NFlow_TUnitedParameters_NYT_NFlow_TQueueSource.md -->

### Dynamic spec

<!-- source: en/flow/generated_docs/NYT_NFlow_TDynamicUnitedParameters_NYT_NFlow_TQueueSource.md -->
<!-- This file is generated by yt/yt/flow/yandex/tools/generate_yson_struct_doc/generate.sh script -->
<!-- Before using doc generation tool check readme: yt/yt/flow/yandex/tools/generate_yson_struct_doc/README.md -->
Source: [yt/yt/flow/library/cpp/common/registry-inl.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/common/registry-inl.h)

#|
|| **Parameter** | **Description** ||
|| `pull_queue_timeout` | **Type**: [TDuration](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#TDuration)
**Default value**: `1m`
Timeout for a read request from the queue. ||
|#


{% cut "**Additional parameters**" %}


#|
|| `unavailable_threshold` | **Type**: [TDuration](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#TDuration)
**Default value**: `5m`
How long the source must be continuously unavailable for the partition to count as stably unavailable. Only time during which the job was running and seeing the failure counts: a gap between restarts is not charged, and any successful answer from the source resets the total. ||
|#


{% endcut %}
<!-- endsource: en/flow/generated_docs/NYT_NFlow_TDynamicUnitedParameters_NYT_NFlow_TQueueSource.md -->

## Writing to a Queue

To write messages to a queue in a [sink](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#sink), send messages with the same schema as the table where you plan to write. You can also configure writing metadata to a special column to pass information about the message’s `event_time` and the stream watermark to queue readers.

There are two sink options: synchronous (`NYT::NFlow::TSyncQueueSink`) and asynchronous (`NYT::NFlow::TAsyncQueueSink`). Writing to a synchronous sink happens in the main transaction of the [epoch](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#epoch). This is efficient, but you can only write to a queue on the main processing cluster. Writing to an asynchronous sink happens after the main epoch transaction, using messages stored in output messages. This is more resource-intensive, but you can write to a queue on any cluster.

### Parameters for synchronous sink specs

#### Static spec

<!-- source: en/flow/generated_docs/NYT_NFlow_TUnitedParameters_NYT_NFlow_TSyncQueueSink.md -->
<!-- This file is generated by yt/yt/flow/yandex/tools/generate_yson_struct_doc/generate.sh script -->
<!-- Before using doc generation tool check readme: yt/yt/flow/yandex/tools/generate_yson_struct_doc/README.md -->
Source: [yt/yt/flow/library/cpp/common/registry-inl.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/common/registry-inl.h)

#|
|| **Parameter** | **Description** ||
|| `queue_path` | **Type**: `NYT::NYPath::TRichYPath`
**Required parameter**
Path to the queue including the cluster. ||
|| `write_flow_queue_meta` | **Type**: `bool`
**Default value**: `false`
Whether to write meta to the queue (watermarks are written in meta). ||
|| `flow_queue_meta_column` | **Type**: `std::string`
**Default value**: `flow_queue_meta`
Column to which to write meta. ||
|| `tablet_index_expression` | **Type**: `std::optional<std::string>`
Verbatim routing mode. A QL expression (builtins only, e.g. `farm_hash`) over the message columns whose value is written to the system column `$tablet_index` verbatim. Mutually exclusive with `tablet_index_routing_hash_expression`; when neither expression is set, routing is off (`$tablet_index` is not written and the tablet is chosen by the driver). Sharded queues are inconvenient to operate (object&rarr;tablet affinity is tightly coupled and a reshard remaps keys), so enable routing only when genuinely needed. Routing is supported only on sync queue sinks; async sinks reject the routing params at spec load (YTFLOW-766). ||
|| `tablet_index_routing_hash_expression` | **Type**: `std::optional<std::string>`
Hash routing mode. A QL expression (builtins only, e.g. `farm_hash`) yielding a `uint64` hash that is reduced to a tablet index by `tablet_index_routing_hash_policy` over `tablet_count`. Mutually exclusive with `tablet_index_expression`. Supported only on sync queue sinks; async sinks reject the routing params (YTFLOW-766). ||
|| `tablet_index_routing_hash_policy` | **Type**: `std::optional<`[NYT::NFlow::EQueueTabletIndexRoutingHashPolicy](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#NYT_NFlow_EQueueTabletIndexRoutingHashPolicy)`>`
The policy for reducing the `tablet_index_routing_hash_expression` hash to a tablet index. Required together with it. `range` &mdash; contiguous equal-width hash ranges (rangeSize = 2^64 / tablet_count), recommended: a consumer range-partitioned by the same key reads only its own tablet. `modulo` &mdash; `hash % tablet_count`; discouraged: Flow computations are range-partitioned by key, so a modulo-partitioned queue forces every reader to read every tablet (a full mesh on read). ||
|| `tablet_count` | **Type**: `std::optional<long>`
The number of tablets used to reduce `tablet_index_routing_hash_expression` to a tablet index. Optional. When unset, the sink periodically re-reads the target queue's `@tablet_count` (from the master cache) and follows a reshard without a restart. When set explicitly, the value is fixed (no queue read) and a reshard is reflected only by changing the config. ||
|| `column_filter` | **Type**: `std::optional<THashSet<std::string>>`
Which columns from the message to write to the queue (default &mdash; all). ||
|#


{% cut "**Additional parameters**" %}


#|
|| `update_partition_count_period` | **Type**: [TDuration](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#TDuration)
**Default value**: `1m`
How often the controller should update the partition count in the queue (the controller changes the computation partition count according to the queue partition count). ||
|| `update_partition_count_retry_min_backoff` | **Type**: [TDuration](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#TDuration)
**Default value**: `1s`
Initial delay before retrying a failed partition count update, subject to jitter. The value is capped by `update_partition_count_period`; subsequent delays increase exponentially. ||
|#


{% endcut %}
<!-- endsource: en/flow/generated_docs/NYT_NFlow_TUnitedParameters_NYT_NFlow_TSyncQueueSink.md -->

#### Dynamic spec

<!-- source: en/flow/generated_docs/NYT_NFlow_TDynamicUnitedParameters_NYT_NFlow_TSyncQueueSink.md -->
<!-- This file is generated by yt/yt/flow/yandex/tools/generate_yson_struct_doc/generate.sh script -->
<!-- Before using doc generation tool check readme: yt/yt/flow/yandex/tools/generate_yson_struct_doc/README.md -->
Source: [yt/yt/flow/library/cpp/common/registry-inl.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/common/registry-inl.h)

**This structure has no main parameters.**


{% cut "**Additional parameters**" %}


#|
|| `flow_queue_meta_heartbeat_period` | **Type**: [TDuration](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#TDuration)
**Default value**: `10s`
How often the controller will write heartbeats with meta containing watermarks to all queue partitions. ||
|#


{% endcut %}
<!-- endsource: en/flow/generated_docs/NYT_NFlow_TDynamicUnitedParameters_NYT_NFlow_TSyncQueueSink.md -->

### Parameters for asynchronous sink specs

#### Static spec

<!-- source: en/flow/generated_docs/NYT_NFlow_TUnitedParameters_NYT_NFlow_TAsyncQueueSink.md -->
<!-- This file is generated by yt/yt/flow/yandex/tools/generate_yson_struct_doc/generate.sh script -->
<!-- Before using doc generation tool check readme: yt/yt/flow/yandex/tools/generate_yson_struct_doc/README.md -->
Source: [yt/yt/flow/library/cpp/common/registry-inl.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/common/registry-inl.h)

#|
|| **Parameter** | **Description** ||
|| `queue_path` | **Type**: `NYT::NYPath::TRichYPath`
**Required parameter**
Path to the queue including the cluster. ||
|| `write_flow_queue_meta` | **Type**: `bool`
**Default value**: `false`
Whether to write meta to the queue (watermarks are written in meta). ||
|| `flow_queue_meta_column` | **Type**: `std::string`
**Default value**: `flow_queue_meta`
Column to which to write meta. ||
|| `producer_path` | **Type**: `NYT::NYPath::TRichYPath`
**Required parameter**
Path to the queue producer with the cluster specified. ||
|| `require_sync_replica` | **Type**: `bool`
**Default value**: `true`
The eponymous parameter when writing to the queue. Whether write to the queue without synchronous replicas is allowed. ||
|| `tablet_index_expression` | **Type**: `std::optional<std::string>`
Verbatim routing mode. A QL expression (builtins only, e.g. `farm_hash`) over the message columns whose value is written to the system column `$tablet_index` verbatim. Mutually exclusive with `tablet_index_routing_hash_expression`; when neither expression is set, routing is off (`$tablet_index` is not written and the tablet is chosen by the driver). Sharded queues are inconvenient to operate (object&rarr;tablet affinity is tightly coupled and a reshard remaps keys), so enable routing only when genuinely needed. Routing is supported only on sync queue sinks; async sinks reject the routing params at spec load (YTFLOW-766). ||
|| `tablet_index_routing_hash_expression` | **Type**: `std::optional<std::string>`
Hash routing mode. A QL expression (builtins only, e.g. `farm_hash`) yielding a `uint64` hash that is reduced to a tablet index by `tablet_index_routing_hash_policy` over `tablet_count`. Mutually exclusive with `tablet_index_expression`. Supported only on sync queue sinks; async sinks reject the routing params (YTFLOW-766). ||
|| `tablet_index_routing_hash_policy` | **Type**: `std::optional<`[NYT::NFlow::EQueueTabletIndexRoutingHashPolicy](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#NYT_NFlow_EQueueTabletIndexRoutingHashPolicy)`>`
The policy for reducing the `tablet_index_routing_hash_expression` hash to a tablet index. Required together with it. `range` &mdash; contiguous equal-width hash ranges (rangeSize = 2^64 / tablet_count), recommended: a consumer range-partitioned by the same key reads only its own tablet. `modulo` &mdash; `hash % tablet_count`; discouraged: Flow computations are range-partitioned by key, so a modulo-partitioned queue forces every reader to read every tablet (a full mesh on read). ||
|| `tablet_count` | **Type**: `std::optional<long>`
The number of tablets used to reduce `tablet_index_routing_hash_expression` to a tablet index. Optional. When unset, the sink periodically re-reads the target queue's `@tablet_count` (from the master cache) and follows a reshard without a restart. When set explicitly, the value is fixed (no queue read) and a reshard is reflected only by changing the config. ||
|| `at_most_once_strategy` | **Type**: `NYT::TIntrusivePtr<`[NYT::NFlow::TAtMostOnceStrategyParameters](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#NYT_NFlow_TAtMostOnceStrategyParameters)`>`
Optional at-most-once delivery strategy. Connector support varies; check the connector documentation before enabling it. ||
|| `column_filter` | **Type**: `std::optional<THashSet<std::string>>`
Which columns from the message to write to the queue (default &mdash; all). ||
|#


{% cut "**Additional parameters**" %}


#|
|| `update_partition_count_period` | **Type**: [TDuration](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#TDuration)
**Default value**: `1m`
How often the controller should update the partition count in the queue (the controller changes the computation partition count according to the queue partition count). ||
|| `update_partition_count_retry_min_backoff` | **Type**: [TDuration](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#TDuration)
**Default value**: `1s`
Initial delay before retrying a failed partition count update, subject to jitter. The value is capped by `update_partition_count_period`; subsequent delays increase exponentially. ||
|#


{% endcut %}
<!-- endsource: en/flow/generated_docs/NYT_NFlow_TUnitedParameters_NYT_NFlow_TAsyncQueueSink.md -->

#### Dynamic spec

<!-- source: en/flow/generated_docs/NYT_NFlow_TDynamicUnitedParameters_NYT_NFlow_TAsyncQueueSink.md -->
<!-- This file is generated by yt/yt/flow/yandex/tools/generate_yson_struct_doc/generate.sh script -->
<!-- Before using doc generation tool check readme: yt/yt/flow/yandex/tools/generate_yson_struct_doc/README.md -->
Source: [yt/yt/flow/library/cpp/common/registry-inl.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/common/registry-inl.h)

#|
|| **Parameter** | **Description** ||
|| `at_most_once_strategy` | **Type**: `NYT::TIntrusivePtr<`[NYT::NFlow::TAtMostOnceStrategyDynamicParameters](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#NYT_NFlow_TAtMostOnceStrategyDynamicParameters)`>`
**Default value**: `{}`
Dynamic parameters for `at_most_once_strategy`. Connector support varies; check the connector documentation before configuring them. ||
|#


{% cut "**Additional parameters**" %}


#|
|| `flow_queue_meta_heartbeat_period` | **Type**: [TDuration](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#TDuration)
**Default value**: `10s`
How often the controller will write heartbeats with meta containing watermarks to all queue partitions. ||
|| `write_period` | **Type**: [TDuration](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#TDuration)
**Default value**: `100ms`
 ||
|| `max_rows_per_write` | **Type**: `long`
**Default value**: `1000`
 ||
|| `max_bytes_per_write` | **Type**: `long`
**Default value**: `1048576`
 ||
|| `backoff_duration` | **Type**: [TDuration](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#TDuration)
**Default value**: `1s`
 ||
|#


{% endcut %}
<!-- endsource: en/flow/generated_docs/NYT_NFlow_TDynamicUnitedParameters_NYT_NFlow_TAsyncQueueSink.md -->


## See also

- [List of Connectors](https://ytsaurus.tech/docs/en/flow/connectors/about.md)
- [Spec and DynamicSpec](https://ytsaurus.tech/docs/en/flow/concepts/spec.md)
<!-- endsource: en/_includes/flow/connectors/queue.md -->