diff --git a/README.md b/README.md
index c25e788..025cce4 100644
--- a/README.md
+++ b/README.md
@@ -113,8 +113,8 @@ Main components:
   and manages the replication position;
 - `DstTable` reads the destination table schema and converts CDC values to its YDB
   types;
-- `Locker` prevents two processes with the same `instance_id` from writing the
-  replica concurrently.
+- `Locker` prevents two processes using the same destination state table and
+  `instance_id` from writing the replica concurrently.
 
 ## Replication algorithm
 
@@ -247,19 +247,33 @@ heartbeats older than the stored position are immediately committed without bein
 applied. If the stored state is not `OK`, automatic continuation is refused: fix
 the cause and recover the state deliberately first.
 
-### Multiple instances
-
-Before reading source topics, the process acquires a lock in the row for its
-`instance_id`. Lock TTL is `2 × max_expected_heartbeat_interval`.
-
-- `multiple_instances_mode: true`: standby processes keep waiting and may take
-  over after the lock is released;
-- `multiple_instances_mode: false`: the process stops waiting after a series of
-  failed checks, performed every 5 seconds.
-
-Processes for independent replication sets sharing one state table need different
-`instance_id` values. Active/standby processes for the same replication use the
-same `instance_id` and identical stream configuration.
+### Feed ownership and multiple pods
+
+A **feed** is a logical replication unit: its streams share one heartbeat quorum,
+checkpoint, and destination transaction. A **pod/process** is a worker that
+acquires ownership of an available feed from the configured `feeds` list.
+`instance_id` identifies persistent replication state and locking; it defaults to
+the feed's `id` and must remain stable across pod replacements.
+
+Multiple pods use the same configuration to run different feeds concurrently.
+Ownership is exclusive per feed: before opening CDC read sessions, a worker
+acquires the lock identified by the destination database, `state_table`, and
+effective `instance_id`. Lock TTL is
+`2 × max_expected_heartbeat_interval`. Only the lock owner reads and applies that
+feed. Scaling the pod count allows more feeds to run concurrently; a single feed's
+streams remain together under one owner, preserving their common quorum and
+atomic destination writes. Independent feeds have separate checkpoints and
+transactions.
+
+Workers without an available feed keep waiting until shutdown and may acquire a
+feed after its owner releases the lock or the lock expires. Set the top-level
+`wait_for_feed: false` to exit after approximately 50 seconds without acquiring
+ownership of any feed.
+
+Workers eligible to own the same feed must use the same destination state table,
+`instance_id`, and feed configuration, including consumers. Independent feeds
+sharing a destination state table need different `instance_id` values and separate
+consumers when reading the same source topics.
 
 ## Preparing YDB
 
@@ -321,76 +335,130 @@ The file is supplied through `-config`; the default is `config.yaml` in the curr
 directory. See the full example in
 [`cmd/aardappel/config.yaml`](cmd/aardappel/config.yaml).
 
+The first line must be `version: 2` to select the cluster/feed configuration format.
+
 Configuration is read once at startup. There is no automatic reload: changing
 connection strings, authentication, streams, queues, monitoring, or any other
 setting requires an application restart.
 
 ```yaml
-src_connection_string: "grpcs://source.example.net:2135/Root/source"
-src_client_balancer: true
-src_oauth2_file: "/run/secrets/source-oauth.json"
-# src_static_token: "..."
-# src_oauth2_endpoint: "https://sts.example.net/oauth2/token/exchange"
-
-dst_connection_string: "grpcs://destination.example.net:2135/Root/destination"
-dst_client_balancer: true
-dst_oauth2_file: "/run/secrets/destination-oauth.json"
-# dst_static_token: "..."
-# dst_oauth2_endpoint: "https://sts.example.net/oauth2/token/exchange"
-
-state_table: "/Root/aardappel_state"
-instance_id: "orders-replica"
-multiple_instances_mode: true
-max_expected_heartbeat_interval: 10
+version: 2
+clusters:
+  - id: source
+    connection_string: "grpcs://source.example.net:2135/Root/source"
+    client_balancer: true
+    oauth2_file: "/run/secrets/source-oauth.json"
+    # static_token: "..."
+    # oauth2_endpoint: "https://sts.example.net/oauth2/token/exchange"
+  - id: destination
+    connection_string: "grpcs://destination.example.net:2135/Root/destination"
+    client_balancer: true
+    oauth2_file: "/run/secrets/destination-oauth.json"
+    # static_token: "..."
+    # oauth2_endpoint: "https://sts.example.net/oauth2/token/exchange"
+
+stream_sets:
+  - id: orders
+    streams:
+      - src_topic: "/Root/orders/orders_cdc"
+        dst_table: "/Root/orders_replica"
+        problem_strategy: "stop"
+        mon_tag: "orders"
+
+feeds:
+  - id: "orders-replica"
+    source: source
+    destination: destination
+    state_table: "/Root/aardappel_state"
+    max_expected_heartbeat_interval: 10
+
+    streams:
+      - ref: orders
+        consumer: "aardappel"
+
+    # Optional:
+    # cmd_queue:
+    #   path: "/Root/aardappel_commands"
+    #   consumer: "aardappel"
+    # dead_letter_queue:
+    #   path: "/Root/aardappel_dlq"
+    # key_filter:
+    #   table_path: "/Root/aardappel_blocked_keys"
+
+wait_for_feed: true
 log_level: "info"
-
-streams:
-  - src_topic: "/Root/orders/orders_cdc"
-    consumer: "aardappel"
-    dst_table: "/Root/orders_replica"
-    problem_strategy: "stop"
-    mon_tag: "orders"
-
 mon_server:
   listen: ":8080"
-
-# Optional:
-# cmd_queue:
-#   path: "/Root/aardappel_commands"
-#   consumer: "aardappel"
-# dead_letter_queue:
-#   path: "/Root/aardappel_dlq"
-# key_filter:
-#   table_path: "/Root/aardappel_blocked_keys"
 ```
 
 ### Top-level settings
 
 | Setting | Required | Purpose |
 |---|---:|---|
-| `src_connection_string` | yes | Source YDB connection string. |
-| `src_client_balancer` | no | `false` enables `SingleConn`, useful when direct node access is unavailable. |
-| `src_oauth2_file` / `src_static_token` | yes | Exactly one source authentication method. |
-| `src_oauth2_endpoint` | no | Overrides the credentials-file token exchange endpoint. |
-| `dst_connection_string` | yes | Destination YDB connection string. |
-| `dst_client_balancer` | no | Equivalent destination setting. |
-| `dst_oauth2_file` / `dst_static_token` | yes | Exactly one destination authentication method. |
-| `dst_oauth2_endpoint` | no | Overrides the destination token exchange endpoint. |
+| `clusters` | yes | Reusable cluster definitions referenced by feeds. |
+| `stream_sets` | no | Reusable stream mappings referenced by `feeds[].streams[].ref`. |
+| `feeds` | yes | Replication feeds. |
+| `wait_for_feed` | no | Defaults to `true`: keep waiting to acquire ownership of an available configured feed. When `false`, exit after approximately 50 seconds without acquiring feed ownership. |
+| `log_level` | no | `debug`, `info`, `warn`, or `error`. |
+| `mon_server` | no | Enables metrics and readiness HTTP endpoints. |
+
+### `clusters[]` settings
+
+Each cluster has a unique `id` and reusable connection and authentication settings.
+
+| Setting | Required | Purpose |
+|---|---:|---|
+| `id` | yes | Unique cluster ID referenced by feeds. |
+| `connection_string` | yes | Cluster YDB connection string. |
+| `client_balancer` | no | `false` enables `SingleConn`, useful when direct node access is unavailable. |
+| `oauth2_file` / `src_static_token` | yes | Exactly one source authentication method. |
+| `oauth2_endpoint` | no | Overrides the credentials-file token exchange endpoint. |
+
+### `stream_sets[]` settings
+
+Each set defines reusable stream mappings without consumers. Its `streams` entries
+use the fields described under `streams[]` below, except that `consumer` must be
+omitted and is supplied by each feed reference.
+
+| Setting | Required | Purpose |
+|---|---:|---|
+| `id` | yes | Unique stream set ID referenced by `feeds[].streams[].ref`. |
+| `streams` | yes | Nonempty list of topic-to-table mappings. |
+
+### `feeds[]` settings
+
+Each feed has a unique `id` and references its source and destination clusters.
+Set `instance_id` to the previous value to preserve replication state. It defaults
+to `id` and may be reused for different destinations. Execution locks use
+`instance_id` in the destination state table. Configuration loading warns when
+different feeds use the same effective `instance_id` and destination cluster ID.
+
+| Setting | Required | Purpose |
+|---|---:|---|
+| `id` | yes | Unique feed identifier used for monitoring. |
+| `instance_id` | no | Replication state, locking, key-filter, and conflict-command identifier; defaults to `id`. |
+| `source` / `destination` | yes | Cluster IDs for each feed. |
 | `state_table` | yes | Destination service table for state and locking. |
-| `instance_id` | yes | Replication identifier and state table row key. |
-| `multiple_instances_mode` | no | Keeps a standby process waiting for the lock. |
 | `commit_offset_mode` | no | Uses `CommitOffset` instead of `Commit` to commit offsets in source CDC topics. |
 | `max_concurrent_offset_commits` | no | Maximum number of concurrent `CommitOffset` requests. |
-| `streams` | yes | CDC topic to destination table mappings. |
+| `streams` | yes | List of inline stream mappings or list of stream set references. |
 | `max_expected_heartbeat_interval` | recommended | Missing-heartbeat warning threshold and lock TTL input. |
-| `log_level` | no | `debug`, `info`, `warn`, or `error`. |
-| `mon_server` | no | Enables metrics and readiness HTTP endpoints. |
 | `cmd_queue` | no | External `skip`/`apply` decisions for out-of-order messages. |
 | `dead_letter_queue` | no | Diagnostic topic for problem transactions. |
 | `key_filter` | no | Table of keys whose changes must not be applied. |
 
+### Stream set references in `feeds[].streams[]`
+
+| Setting | Required | Purpose |
+|---|---:|---|
+| `ref` | yes | ID of a declared stream set. |
+| `consumer` | yes | Consumer to use for every topic in the referenced set. |
+
 ### `streams[]` settings
 
+These fields describe inline feed streams and mappings inside `stream_sets[].streams`.
+The `consumer` field is required for inline feed streams and must be omitted in sets.
+
 | Setting | Required | Purpose |
 |---|---:|---|
 | `src_topic` | yes | Source CDC topic path. |
@@ -399,9 +467,8 @@ mon_server:
 | `problem_strategy` | no | `stop` (default) or `continue`. |
 | `mon_tag` | no | `stream_tag` label; defaults to `src_topic`. |
 
-A consumer must be exclusive to one concurrently running Aardappel replication
-group. The same consumer name is fine in different CDC topics, but independent
-instances must not read the same consumer of the same topic.
+A consumer must be exclusive to one feed and the workers eligible to own it. The same consumer name is fine in different CDC topics, but independent
+feeds must not read the same consumer of the same topic.
 
 Stream order is not stored in the state table; only the common
 `(step_id, tx_id)` position is stored. After restart, configuration is reread and
@@ -415,7 +482,7 @@ new `instance_id`, consumer/state, or preliminary data alignment.
 
 ## Adding a new stream
 
-A new stream can be added to an existing replication instance, for example when a
+A new stream can be added to an existing feed, for example when a
 new source table must be replicated consistently with existing tables. It joins
 the same heartbeat quorum, and its changes are applied in the same destination
 transaction as changes from other streams.
@@ -424,7 +491,7 @@ The main task when adding a stream is to **align its position with the existing
 replication**. There are two independent position layers:
 
 - `step_id` and `tx_id` in `state_table` form one global virtual timestamp for all
-  streams of the instance. Aardappel uses it to filter out messages that it
+  streams of the feed. Aardappel uses it to filter out messages that it
   considers already processed;
 - each CDC consumer stores committed offsets separately for every topic partition.
   These offsets determine where physical reading of each stream resumes.
@@ -467,11 +534,11 @@ Use this procedure when the new table and CDC already contain data, but the enti
 history before the first common quorum is small enough for one destination
 transaction:
 
-1. Stop the active process and every standby process with the same `instance_id`.
+1. Stop all workers eligible to own the feed.
 2. Back up the current state row and record committed offsets for every consumer
    partition.
 3. Add the new stream to the configuration.
-4. Reset the global timestamp for this instance to `(0, 0)` so that Aardappel does
+4. Reset the global timestamp for this feed to `(0, 0)` so that Aardappel does
    not filter the new stream's historical messages using the old checkpoint:
 
    ```sql
@@ -492,11 +559,11 @@ at a common heartbeat quorum, after which replication continues normally.
 Use this procedure when the new CDC contains an initial scan that is too large for
 one destination request or transaction:
 
-1. Stop the active process and every standby process with the same `instance_id`.
+1. Stop all workers eligible to own the feed.
 2. Back up the current state row and record committed offsets for every consumer
    partition.
 3. Add the new stream to the configuration.
-4. Reset the global timestamp to `(0, 0)` and switch the instance to
+4. Reset the global timestamp to `(0, 0)` and switch the feed to
    `INITIAL_SCAN`:
 
    ```sql
@@ -515,24 +582,23 @@ position, switches `stage` to `RUN` automatically, and resumes normal replicatio
 The existing streams continue from their committed consumer offsets; the new
 stream reads and applies its initial-scan history and catches up with them.
 
-### Alternative: preliminary replication through a separate instance
+### Alternative: preliminary replication through a separate feed
 
 For a large table, the new stream can first be replicated through a temporary,
-separate Aardappel instance:
+separate feed running in its own process or pod group:
 
 1. Create a separate `instance_id`, state row, and consumer for the new stream.
-2. Replicate the new table until the temporary instance catches up.
-3. Stop both the main and temporary replication instances.
+2. Replicate the new table until the temporary feed catches up.
+3. Stop all eligible worker processes for both the main and temporary feeds.
 4. Add the new stream to the main configuration and perform the alignment from
-   Scenario 2: reset the main global timestamp to `(0, 0)` and start the main
-   instance.
+   Scenario 2: reset the main global timestamp to `(0, 0)` and restart the main feed.
 5. Reuse or align the consumer offset at the confirmed cutover point so that the
-   main instance continues from the data already applied by the temporary
+   main feed continues from the data already applied by the temporary
    replication.
 6. After checking the common quorum, lag, and destination data, remove the
-   temporary replication instance.
+   temporary feed and its processes.
 
-Never let the temporary and main instances read the same consumer concurrently.
+Never let the temporary and main feeds read the same consumer concurrently.
 
 ### Transition period and safety
 
@@ -549,20 +615,21 @@ completion using heartbeat/quorum, replication lag, logs, and an application-lev
 data comparison.
 
 > **Do not concurrently use one consumer of one CDC topic from independent
-> Aardappel instances.** YDB distributes partitions between read sessions: one
+> feeds.** YDB distributes partitions between read sessions: one
 > process receives one subset and another receives the rest. Each
 > `HeartBeatTracker` expects all partitions, so neither process can form its
 > expected full quorum correctly. Create separate consumers for independent
-> instances. Active/standby processes with the same `instance_id` may share a
-> consumer because the distributed lock permits only one process to read.
+> feeds. Workers eligible to own the same feed may share a consumer
+> because they use the same destination state row and distributed lock, which
+> permits only one process to read.
 
 ## Authentication
 
-Source and destination authentication are configured independently. Set
-**exactly one** method for each side:
+Authentication is configured independently for each cluster. Set
+**exactly one** method per cluster:
 
-- `*_static_token`: a final YDB access token without OAuth2 exchange;
-- `*_oauth2_file`: a JSON file describing OAuth2 token exchange.
+- `static_token`: a final YDB access token without OAuth2 exchange;
+- `oauth2_file`: a JSON file describing OAuth2 token exchange;
 
 Two credential-file formats are supported:
 
@@ -572,9 +639,7 @@ Two credential-file formats are supported:
    files or inline values. This is convenient for projected Kubernetes service
    account tokens.
 
-`src_oauth2_endpoint` and `dst_oauth2_endpoint` override the endpoint in their
-respective credentials file. See [`internal/auth/README.md`](internal/auth/README.md)
-for complete field descriptions and JSON examples.
+Each cluster’s `oauth2_endpoint` overrides the endpoint in its credentials file.
 
 ## Problem message handling
 
@@ -584,7 +649,7 @@ a lower position violates the expected stream order.
 ```mermaid
 flowchart TD
     O[Out-of-order change received] --> C{Is cmd_queue configured?}
-    C -->|yes| F{Command found for<br/>instance + topic + key + ts?}
+    C -->|yes| F{Command found for<br/>instance_id + topic + key + ts?}
     F -->|apply| A[Apply change]
     F -->|skip| S[Skip and commit offset]
     F -->|no| P{problem_strategy}
@@ -641,24 +706,26 @@ table path. Do not construct it manually without the same serializer.
 
 ## Monitoring
 
+All application metrics include `feed_id` from the configured feed `id`. Go runtime collector metrics remain process-scoped.
+
 When `mon_server.listen` is configured, the service exposes:
 
 - `GET /metrics`: Prometheus metrics;
 - `GET /readyz`: `200` after source topics and destination tables have been
   checked successfully, otherwise `503`.
 
-Multiple instances on one host need different listen ports.
+Multiple processes sharing one network namespace need different listen ports.
 
 | Metric | Type | Meaning |
 |---|---|---|
-| `modifications_count` | counter | Applied modifications. |
-| `modifications_count_per_table{stream_tag}` | counter | Modifications per stream. |
-| `commit_latency` | histogram | Destination commit duration in seconds. |
-| `quorum_waiting_latency` | histogram | Full heartbeat quorum wait in seconds. |
-| `request_size_bytes` | counter | Estimated cumulative YQL request size. |
-| `replication_lag_estimation` | gauge | Local time minus quorum `step`, seconds. |
-| `topic_without_hb{stream_tag}` | gauge | `1` when a heartbeat is late. |
-| `go_heap_allocated` | gauge | Current heap size in bytes. |
+| `modifications_count{feed_id}` | counter | Applied modifications. |
+| `modifications_count_per_table{feed_id,stream_tag}` | counter | Modifications per stream. |
+| `commit_latency{feed_id}` | histogram | Destination commit duration in seconds. |
+| `quorum_waiting_latency{feed_id}` | histogram | Full heartbeat quorum wait in seconds. |
+| `request_size_bytes{feed_id}` | counter | Estimated cumulative YQL request size. |
+| `replication_lag_estimation{feed_id}` | gauge | Local time minus quorum `step`, seconds. |
+| `topic_without_hb{feed_id,stream_tag}` | gauge | `1` when a heartbeat is late. |
+| `go_heap_allocated{feed_id}` | gauge | Current heap size in bytes. |
 
 `max_expected_heartbeat_interval` should exceed the changefeed
 `RESOLVED_TIMESTAMPS` interval. Exceeding it emits a warning and sets
@@ -688,11 +755,10 @@ context and releasing the lock.
 - Replication is one-way: source YDB → destination YDB.
 - Destination tables and consumers must exist before startup; only `state_table`
   is created automatically.
-- To preserve multi-table consistency, list all CDC streams in one Aardappel
-  instance.
+- To preserve multi-table consistency, put all required CDC streams in the same feed.
 - One consumer of one CDC topic must not be read concurrently by independent
-  Aardappel instances; YDB would split partitions between them. Use one consumer
-  per instance, except for active/standby protected by one lock and `instance_id`.
+  feeds; YDB would split partitions between them. Use one consumer
+  per feed; workers for that feed share it under the same lock.
 - Progress is limited by the slowest partition: no new quorum can form without its
   heartbeat.
 - Reordering unchanged `src_topic → dst_table` pairs is safe after restart.
@@ -821,8 +887,8 @@ YDB SDK назначает внутри неё отдельное чтение 
 - `Processor` выбирает согласованный batch, строит YQL для всех destination-таблиц
   и управляет позицией репликации;
 - `DstTable` читает схему целевой таблицы и преобразует CDC-значения в её YDB-типы;
-- `Locker` не позволяет двум процессам с одним `instance_id` одновременно писать
-  реплику.
+- `Locker` не позволяет двум процессам с одной destination state table и одним
+  `instance_id` одновременно писать реплику.
 
 ## Алгоритм репликации
 
@@ -954,19 +1020,32 @@ heartbeat с позицией меньше сохранённой сразу п
 не выполняется: сначала требуется устранить причину и осознанно восстановить
 состояние.
 
-### Несколько экземпляров
-
-Перед чтением source-топиков процесс получает lock в строке своего `instance_id`.
-TTL lock равен `2 × max_expected_heartbeat_interval`.
-
-- `multiple_instances_mode: true` — standby-экземпляры продолжают ждать lock и
-  могут подхватить работу после его освобождения;
-- `multiple_instances_mode: false` — процесс прекращает ожидание после серии
-  неудачных проверок (проверка выполняется раз в 5 секунд).
-
-У процессов, реплицирующих независимые наборы данных через одну state table,
-должны быть разные `instance_id`. Для active/standby одной репликации, наоборот,
-используется одинаковый `instance_id` и одинаковая конфигурация потоков.
+### Владение feed и несколько pod
+
+**Feed** — логическая единица репликации: его streams используют общий heartbeat
+quorum, checkpoint и destination-транзакцию. **Pod/процесс** — worker, получающий
+владение доступным feed из настроенного списка `feeds`. `instance_id`
+идентифицирует постоянное состояние репликации и lock; по умолчанию он равен `id`
+feed и должен сохраняться при замене pod.
+
+Несколько pod используют общую конфигурацию и выполняют разные feeds параллельно.
+Владение эксклюзивно для каждого feed: перед открытием CDC read sessions worker
+получает lock, определяемый destination-базой, `state_table` и эффективным
+`instance_id`. TTL lock равен `2 × max_expected_heartbeat_interval`.
+Только владелец lock читает и применяет этот feed. Увеличение числа pod позволяет
+выполнять больше feeds параллельно; streams одного feed остаются у одного
+владельца, сохраняя общий quorum и атомарную запись в destination. У независимых
+feeds отдельные checkpoints и транзакции.
+
+Workers, которым не достался доступный feed, продолжают ждать до завершения и
+могут получить feed после освобождения lock его владельцем или истечения срока
+lock. Параметр верхнего уровня `wait_for_feed: false` завершает процесс примерно
+через 50 секунд, если не удалось получить владение ни одним feed.
+
+Workers, которые могут владеть одним и тем же feed, должны использовать одинаковые
+destination state table, `instance_id` и конфигурацию feed, включая consumers.
+Независимым feeds с общей destination state table нужны разные `instance_id` и
+отдельные consumers при чтении одних и тех же source-топиков.
 
 ## Подготовка YDB
 
@@ -1033,76 +1112,130 @@ source и только после этого начинайте записыва
 текущего каталога. Полный пример находится в
 [`cmd/aardappel/config.yaml`](cmd/aardappel/config.yaml).
 
+Первая строка должна быть `version: 2`, чтобы выбрать формат конфигурации с кластерами и feeds.
+
 Конфигурация читается один раз при старте. Автоматического reload файла нет:
 любое изменение connection strings, авторизации, streams, очередей, мониторинга
 или других параметров требует перезапуска приложения.
 
 ```yaml
-src_connection_string: "grpcs://source.example.net:2135/Root/source"
-src_client_balancer: true
-src_oauth2_file: "/run/secrets/source-oauth.json"
-# src_static_token: "..."               # альтернатива src_oauth2_file
-# src_oauth2_endpoint: "https://sts.example.net/oauth2/token/exchange"
-
-dst_connection_string: "grpcs://destination.example.net:2135/Root/destination"
-dst_client_balancer: true
-dst_oauth2_file: "/run/secrets/destination-oauth.json"
-# dst_static_token: "..."               # альтернатива dst_oauth2_file
-# dst_oauth2_endpoint: "https://sts.example.net/oauth2/token/exchange"
-
-state_table: "/Root/aardappel_state"
-instance_id: "orders-replica"
-multiple_instances_mode: true
-max_expected_heartbeat_interval: 10
+version: 2
+clusters:
+  - id: source
+    connection_string: "grpcs://source.example.net:2135/Root/source"
+    client_balancer: true
+    oauth2_file: "/run/secrets/source-oauth.json"
+    # static_token: "..."               # альтернатива oauth2_file
+    # oauth2_endpoint: "https://sts.example.net/oauth2/token/exchange"
+  - id: destination
+    connection_string: "grpcs://destination.example.net:2135/Root/destination"
+    client_balancer: true
+    oauth2_file: "/run/secrets/destination-oauth.json"
+    # static_token: "..."               # альтернатива oauth2_file
+    # oauth2_endpoint: "https://sts.example.net/oauth2/token/exchange"
+
+stream_sets:
+  - id: orders
+    streams:
+      - src_topic: "/Root/orders/orders_cdc"
+        dst_table: "/Root/orders_replica"
+        problem_strategy: "stop"
+        mon_tag: "orders"
+
+feeds:
+  - id: "orders-replica"
+    source: source
+    destination: destination
+    state_table: "/Root/aardappel_state"
+    max_expected_heartbeat_interval: 10
+
+    streams:
+      - ref: orders
+        consumer: "aardappel"
+
+    # Опционально:
+    # cmd_queue:
+    #   path: "/Root/aardappel_commands"
+    #   consumer: "aardappel"
+    # dead_letter_queue:
+    #   path: "/Root/aardappel_dlq"
+    # key_filter:
+    #   table_path: "/Root/aardappel_blocked_keys"
+
+wait_for_feed: true
 log_level: "info"
-
-streams:
-  - src_topic: "/Root/orders/orders_cdc"
-    consumer: "aardappel"
-    dst_table: "/Root/orders_replica"
-    problem_strategy: "stop"
-    mon_tag: "orders"
-
 mon_server:
   listen: ":8080"
-
-# Опционально:
-# cmd_queue:
-#   path: "/Root/aardappel_commands"
-#   consumer: "aardappel"
-# dead_letter_queue:
-#   path: "/Root/aardappel_dlq"
-# key_filter:
-#   table_path: "/Root/aardappel_blocked_keys"
 ```
 
 ### Параметры верхнего уровня
 
 | Параметр | Обязателен | Назначение |
 |---|---:|---|
-| `src_connection_string` | да | YDB connection string исходной базы. |
-| `src_client_balancer` | нет | `false` включает `SingleConn`; полезно, когда прямой доступ к узлам невозможен. |
-| `src_oauth2_file` / `src_static_token` | да | Ровно один способ авторизации source. |
-| `src_oauth2_endpoint` | нет | Переопределяет endpoint обмена токенов из credentials-файла. |
-| `dst_connection_string` | да | YDB connection string целевой базы. |
-| `dst_client_balancer` | нет | Аналогичный параметр для destination. |
-| `dst_oauth2_file` / `dst_static_token` | да | Ровно один способ авторизации destination. |
-| `dst_oauth2_endpoint` | нет | Переопределяет endpoint обмена токенов destination. |
+| `clusters` | да | Определения кластеров, на которые ссылаются feeds. |
+| `stream_sets` | нет | Наборы streams, на которые ссылается `feeds[].streams[].ref`. |
+| `feeds` | да | Потоки репликации. |
+| `wait_for_feed` | нет | По умолчанию `true`: ждать получения владения доступным feed из конфигурации. При `false` завершить процесс примерно через 50 секунд, если не удалось получить владение feed. |
+| `log_level` | нет | `debug`, `info`, `warn` или `error`. |
+| `mon_server` | нет | Включает HTTP endpoints метрик и readiness. |
+
+### Параметры `clusters[]`
+
+Каждый кластер имеет уникальный `id` и параметры подключения и авторизации.
+
+| Параметр | Обязателен | Назначение |
+|---|---:|---|
+| `id` | да | Уникальный ID кластера. |
+| `connection_string` | да | YDB connection string кластера. |
+| `client_balancer` | нет | `false` включает `SingleConn`; полезно, когда прямой доступ к узлам невозможен. |
+| `oauth2_file` / `dst_static_token` | да | Ровно один способ авторизации кластера. |
+| `oauth2_endpoint` | нет | Переопределяет endpoint обмена токенов кластера. |
+
+### Параметры `stream_sets[]`
+
+| Параметр | Обязателен | Назначение |
+|---|---:|---|
+| `id` | да | Уникальный ID набора, используемый в `feeds[].streams[].ref`. |
+| `streams` | да | Непустой список соответствий CDC-топиков и destination-таблиц. |
+
+### Параметры `feeds[]`
+
+Каждый feed имеет уникальный `id` и ссылается на исходный и целевой кластеры.
+Для сохранения прогресса задайте прежнее значение `instance_id`. По умолчанию
+используется `id`; для разных destinations можно использовать одинаковый
+`instance_id`. Блокировка использует `instance_id` в таблице состояния destination.
+При загрузке конфигурации выводится предупреждение, если разные feeds используют
+одинаковые эффективный `instance_id` и ID destination-кластера.
+
+| Параметр | Обязателен | Назначение |
+|---|---:|---|
+| `id` | да | Уникальный идентификатор feed для мониторинга. |
+| `instance_id` | нет | Идентификатор состояния, блокировки, key filter и команд конфликтов; по умолчанию `id`. |
+| `source` / `destination` | да | ID кластеров для feed. |
 | `state_table` | да | Путь сервисной таблицы состояния и lock в destination. |
-| `instance_id` | да | Идентификатор репликации и ключ строки в state table. |
-| `multiple_instances_mode` | нет | Разрешает standby-процессу продолжать ожидание lock. |
 | `commit_offset_mode` | нет | Использует `CommitOffset` вместо `Commit` для коммита оффсетов исходных CDC-топиков. |
 | `max_concurrent_offset_commits` | нет | Максимальное число одновременных запросов `CommitOffset`. |
-| `streams` | да | Список соответствий CDC-топиков и destination-таблиц. |
+| `streams` | да | Список inline streams или список ссылок на наборы streams. |
 | `max_expected_heartbeat_interval` | рекомендуется | Порог предупреждения об отсутствующем heartbeat; также определяет TTL lock. |
-| `log_level` | нет | `debug`, `info`, `warn` или `error`. |
-| `mon_server` | нет | Включает HTTP endpoints метрик и readiness. |
 | `cmd_queue` | нет | Топик внешних решений `skip`/`apply` для out-of-order сообщений. |
 | `dead_letter_queue` | нет | Топик для диагностических сообщений о проблемных транзакциях. |
 | `key_filter` | нет | Таблица ключей, изменения которых не нужно применять. |
 
+### Ссылки на наборы в `feeds[].streams[]`
+
+Ссылки задаются списком, даже для одного набора. Consumer ссылки применяется ко
+всем streams этого набора. Ссылка не может содержать поля inline stream.
+
+| Параметр | Обязателен | Назначение |
+|---|---:|---|
+| `ref` | да | ID объявленного набора streams. |
+| `consumer` | да | Consumer для каждого топика в указанном наборе. |
+
 ### Параметры `streams[]`
 
+Эти поля описывают inline streams feed и соответствия в `stream_sets[].streams`.
+Поле `consumer` обязательно для inline streams и должно быть опущено в наборах.
+
 | Параметр | Обязателен | Назначение |
 |---|---:|---|
 | `src_topic` | да | Путь к source CDC-топику. |
@@ -1113,7 +1246,7 @@ mon_server:
 
 Consumer должен быть эксклюзивен для одного одновременно работающего контура
 Aardappel. Одинаковое имя consumer допустимо в разных CDC-топиках, но один и тот
-же consumer одного топика нельзя совместно читать независимыми экземплярами.
+же consumer одного топика нельзя совместно читать независимыми feeds.
 
 Порядок `streams` не сохраняется в state table: там хранится только общая позиция
 `(step_id, tx_id)`. После рестарта конфигурация перечитывается, и `TableId` заново
@@ -1128,7 +1261,7 @@ consumer/state или предварительное выравнивание д
 
 ## Добавление нового stream
 
-Новый stream можно добавить в существующий экземпляр репликации, например когда
+Новый stream можно добавить в существующий feed, например когда
 появилась ещё одна source-таблица, которую нужно согласованно реплицировать вместе
 с уже настроенными таблицами. Новый stream войдёт в тот же heartbeat quorum, а его
 изменения будут применяться в одной destination-транзакции с изменениями остальных
@@ -1138,7 +1271,7 @@ streams.
 репликацией**. При этом существуют два независимых уровня позиции:
 
 - `step_id` и `tx_id` в `state_table` образуют один глобальный virtual timestamp
-  всех streams экземпляра. По нему Aardappel фильтрует сообщения, которые считает
+  всех streams feed. По нему Aardappel фильтрует сообщения, которые считает
   уже обработанными;
 - каждый CDC consumer отдельно хранит committed offsets всех партиций топика. Они
   определяют, с какого физического сообщения продолжится чтение каждого stream.
@@ -1179,10 +1312,10 @@ flowchart TD
 Этот вариант подходит, если новая таблица и CDC уже содержат данные, но вся история
 до первого общего quorum достаточно мала для одной destination-транзакции:
 
-1. Остановите активный процесс и все standby-процессы с тем же `instance_id`.
+1. Остановите все workers, которые могут владеть этим feed.
 2. Сохраните текущую строку state и committed offsets всех партиций consumers.
 3. Добавьте новый stream в конфигурацию.
-4. Сбросьте глобальный timestamp экземпляра в `(0, 0)`, чтобы Aardappel не
+4. Сбросьте глобальный timestamp feed в `(0, 0)`, чтобы Aardappel не
    отфильтровал исторические сообщения нового stream по старому checkpoint:
 
    ```sql
@@ -1204,10 +1337,10 @@ streams выровняются на общем heartbeat quorum, после че
 Этот вариант нужен, если новый CDC содержит initial scan, который не помещается в
 один destination-запрос или транзакцию:
 
-1. Остановите активный процесс и все standby-процессы с тем же `instance_id`.
+1. Остановите все workers, которые могут владеть этим feed.
 2. Сохраните текущую строку state и committed offsets всех партиций consumers.
 3. Добавьте новый stream в конфигурацию.
-4. Сбросьте глобальный timestamp в `(0, 0)` и переключите экземпляр в
+4. Сбросьте глобальный timestamp в `(0, 0)` и переключите feed в
    `INITIAL_SCAN`:
 
    ```sql
@@ -1226,25 +1359,25 @@ streams выровняются на общем heartbeat quorum, после че
 Существующие streams продолжат чтение со своих committed consumer offsets, а новый
 stream прочитает и применит initial-scan историю и догонит их.
 
-### Альтернатива: предварительная репликация отдельным экземпляром
+### Альтернатива: предварительная репликация отдельным feed
 
-Большую новую таблицу можно сначала реплицировать временным отдельным экземпляром
-Aardappel:
+Большую новую таблицу можно сначала реплицировать временным отдельным feed
+в собственном процессе или группе pod:
 
 1. Создайте для нового stream отдельные `instance_id`, строку state и consumer.
-2. Реплицируйте новую таблицу до тех пор, пока временный экземпляр не догонит
+2. Реплицируйте новую таблицу до тех пор, пока временный feed не догонит
    source.
-3. Остановите основной и временный экземпляры репликации.
+3. Остановите процессы основного и временного feeds.
 4. Добавьте новый stream в основную конфигурацию и выполните выравнивание из
-   сценария 2: сбросьте глобальный timestamp основного экземпляра в `(0, 0)` и
+   сценария 2: сбросьте глобальный timestamp основного feed в `(0, 0)` и
    запустите его.
 5. Переиспользуйте либо выровняйте consumer offset по подтверждённой точке
-   переключения, чтобы основной экземпляр продолжил с данных, уже применённых
+   переключения, чтобы основной feed продолжил с данных, уже применённых
    временной репликацией.
-6. После проверки общего quorum, lag и destination-данных удалите временную
-   репликацию.
+6. После проверки общего quorum, lag и destination-данных удалите временный
+   feed и его процессы.
 
-Нельзя допускать, чтобы временный и основной экземпляры одновременно читали один
+Нельзя допускать, чтобы временный и основной feeds одновременно читали один
 consumer.
 
 ### Переходный период и безопасность
@@ -1263,20 +1396,21 @@ destination-таблица может быть заполнена только 
 сверке данных.
 
 > **Нельзя одновременно использовать одного consumer одного CDC-топика в разных
-> независимо работающих экземплярах Aardappel.** YDB распределит партиции между
+> независимых feeds.** YDB распределит партиции между
 > read sessions: один процесс получит одну часть партиций, другой — оставшуюся.
 > Каждый `HeartBeatTracker` при этом ожидает heartbeat от всех партиций, поэтому
-> полный quorum не сформируется корректно. Для независимых экземпляров создавайте
-> разные consumers. Active/standby с одним `instance_id` могут использовать общий
-> consumer, поскольку distributed lock допускает к чтению только один процесс.
+> полный quorum не сформируется корректно. Для независимых feeds создавайте
+> разные consumers. Workers одного feed могут использовать общий
+> consumer, поскольку используют одну строку состояния destination и общий
+> distributed lock, допускающий к чтению только один процесс.
 
 ## Авторизация
 
-Source и destination настраиваются независимо. Для каждой стороны необходимо
-задать **ровно один** вариант:
+Авторизация настраивается независимо для каждого кластера. Для каждого кластера
+необходимо задать **ровно один** вариант:
 
-- `*_static_token` — готовый YDB access token без OAuth2 exchange;
-- `*_oauth2_file` — JSON-файл параметров OAuth2 token exchange.
+- `static_token` — готовый YDB access token без OAuth2 exchange;
+- `oauth2_file` — JSON-файл параметров OAuth2 token exchange;
 
 Поддерживаются два формата credentials-файла:
 
@@ -1286,8 +1420,7 @@ Source и destination настраиваются независимо. Для к
    отдельного файла или задать значением. Это удобно для projected service account
    token в Kubernetes.
 
-Параметры `src_oauth2_endpoint` и `dst_oauth2_endpoint` имеют приоритет над
-endpoint внутри соответствующего credentials-файла.
+Параметр `oauth2_endpoint` кластера имеет приоритет над endpoint внутри его credentials-файла.
 
 Полное описание форматов, полей и примеры JSON находятся в
 [`internal/auth/README.md`](internal/auth/README.md).
@@ -1300,7 +1433,7 @@ Reader отслеживает последний heartbeat каждой парт
 ```mermaid
 flowchart TD
     O[Получено out-of-order изменение] --> C{Настроен cmd_queue?}
-    C -->|да| F{Найдена команда для<br/>instance + topic + key + ts?}
+    C -->|да| F{Найдена команда для<br/>instance_id + topic + key + ts?}
     F -->|apply| A[Применить изменение]
     F -->|skip| S[Пропустить и commit offset]
     F -->|нет| P{problem_strategy}
@@ -1343,7 +1476,7 @@ flowchart TD
 попадают в YQL destination-транзакции.
 
 Таблица фильтра создаётся оператором заранее и должна поддерживать чтение диапазона
-по `instance_id` и запись полей:
+по колонке `instance_id` (эффективный ID состояния feed) и запись полей:
 
 ```sql
 CREATE TABLE `/Root/aardappel_blocked_keys` (
@@ -1358,26 +1491,28 @@ CREATE TABLE `/Root/aardappel_blocked_keys` (
 
 ## Мониторинг
 
+Все метрики приложения содержат `feed_id` из поля `id` feed. Метрики Go runtime collector относятся к процессу.
+
 Если настроен `mon_server.listen`, сервис публикует:
 
 - `GET /metrics` — Prometheus metrics;
 - `GET /readyz` — `200` после успешной проверки source-топиков и destination-таблиц,
   до этого `503`.
 
-При нескольких экземплярах на одном адресе каждому нужен отдельный listen port.
+При нескольких процессах в одном сетевом пространстве каждому нужен отдельный listen port.
 
 Основные прикладные метрики:
 
 | Метрика | Тип | Значение |
 |---|---|---|
-| `modifications_count` | counter | Число применённых модификаций. |
-| `modifications_count_per_table{stream_tag}` | counter | Модификации по stream. |
-| `commit_latency` | histogram | Время commit destination-транзакции, секунды. |
-| `quorum_waiting_latency` | histogram | Ожидание полного heartbeat quorum, секунды. |
-| `request_size_bytes` | counter | Оценочный суммарный объём YQL-запросов. |
-| `replication_lag_estimation` | gauge | Разница локального времени и `step` quorum, секунды. |
-| `topic_without_hb{stream_tag}` | gauge | `1`, если heartbeat не пришёл в ожидаемый интервал. |
-| `go_heap_allocated` | gauge | Текущий объём heap, байты. |
+| `modifications_count{feed_id}` | counter | Число применённых модификаций. |
+| `modifications_count_per_table{feed_id,stream_tag}` | counter | Модификации по stream. |
+| `commit_latency{feed_id}` | histogram | Время commit destination-транзакции, секунды. |
+| `quorum_waiting_latency{feed_id}` | histogram | Ожидание полного heartbeat quorum, секунды. |
+| `request_size_bytes{feed_id}` | counter | Оценочный суммарный объём YQL-запросов. |
+| `replication_lag_estimation{feed_id}` | gauge | Разница локального времени и `step` quorum, секунды. |
+| `topic_without_hb{feed_id,stream_tag}` | gauge | `1`, если heartbeat не пришёл в ожидаемый интервал. |
+| `go_heap_allocated{feed_id}` | gauge | Текущий объём heap, байты. |
 
 `max_expected_heartbeat_interval` должен быть больше интервала
 `RESOLVED_TIMESTAMPS` changefeed. Его превышение создаёт warning и выставляет
@@ -1407,11 +1542,11 @@ context и освобождая lock.
 - Destination-таблицы и consumer создаются до запуска; автоматически создаётся
   только `state_table`.
 - Для согласованности нескольких таблиц все их CDC streams должны быть перечислены
-  в одном экземпляре Aardappel.
+  в одном feed.
 - Один consumer одного CDC-топика нельзя одновременно читать из разных независимо
-  работающих экземпляров Aardappel: YDB разделит партиции между ними. Используйте
-  отдельный consumer на экземпляр; исключение — защищённая общим lock пара
-  active/standby с одинаковым `instance_id`.
+  работающих feeds: YDB разделит партиции между ними. Используйте
+  отдельный consumer на feed; его workers используют общий consumer
+  под защитой одного lock.
 - Прогресс ограничен самой медленной партицией: без heartbeat от неё новый quorum
   сформировать нельзя.
 - Перестановка неизменных пар `src_topic → dst_table` допустима после рестарта.
diff --git a/cmd/aardappel/config.yaml b/cmd/aardappel/config.yaml
index 598ed16..a7ffa98 100644
--- a/cmd/aardappel/config.yaml
+++ b/cmd/aardappel/config.yaml
@@ -1,68 +1,83 @@
-## Connection string to the source cluster: grpcs://<ENDPOINT>/<DATABASE>
-src_connection_string: "grpcs://shared-1-grpc.ydb-global.svc.cluster.local:2135/testing-global/aardappel-test"
-## Whether aardappel use client balancer or not.
-## Set to false if direct connection between client and ydb nodes is not possible (k8s)
-## Set to true otherwise
-src_client_balancer: false
-## Source authorizatrion one of:
-##   src_oauth2_file - path to the oauth2 file
-##   src_static_token - YDB static token
-src_oauth2_file: "/opt/aardappel/secrets/src-sa/key.json"
-## OAuth2 token exchange endpoint for src credentials, overrides token-endpoint from src_oauth2_file when set.
-src_oauth2_endpoint: "https://tokens.svc.cluster.local:443/oauth2/token/exchange"
+version: 2
+## Reusable cluster definitions. IDs are referenced by feeds.
+clusters:
+  - id: source
+    ## Connection string to the source cluster: grpcs://<ENDPOINT>/<DATABASE>
+    connection_string: "grpcs://shared-1-grpc.ydb-global.svc.cluster.local:2135/testing-global/aardappel-test"
+    ## Whether aardappel use client balancer or not.
+    ## Set to false if direct connection between client and ydb nodes is not possible (k8s)
+    ## Set to true otherwise
+    client_balancer: false
+    ## Cluster authentication: exactly one of:
+    ##   oauth2_file - path to the oauth2 file
+    ##   static_token - YDB static token
+    oauth2_file: "/opt/aardappel/secrets/src-sa/key.json"
+    ## Overrides the endpoint from the OAuth2 credentials file when set.
+    oauth2_endpoint: "https://tokens.svc.cluster.local:443/oauth2/token/exchange"
 
-## Connection string to the destination cluster: grpcs://<ENDPOINT>/<DATABASE>
-dst_connection_string: "grpcs://shared-1-grpc.ydb-global.svc.cluster.local:2135/testing-global/aardappel-test"
-## Whether aardappel use client balancer or not.
-## Set to false if direct connection between client and ydb nodes is not possible (k8s)
-## Set to true otherwise
-dst_client_balancer: false
-## Destination authorizatrion one of:
-##   dst_oauth2_file - path to the oauth2 file
-##   dst_static_token - YDB static token
-dst_oauth2_file: "/opt/aardappel/secrets/dst-sa/key.json"
-## OAuth2 token exchange endpoint for dst credentials, overrides token-endpoint from dst_oauth2_file when set.
-dst_oauth2_endpoint: "https://tokens.svc.cluster.local:443/oauth2/token/exchange"
+  - id: destination
+    ## Connection string to the destination cluster: grpcs://<ENDPOINT>/<DATABASE>
+    connection_string: "grpcs://shared-1-grpc.ydb-global.svc.cluster.local:2135/testing-global/aardappel-test"
+    ## Whether aardappel use client balancer or not.
+    ## Set to false if direct connection between client and ydb nodes is not possible (k8s)
+    ## Set to true otherwise
+    client_balancer: false
+    ## Cluster authentication: exactly one of:
+    ##   oauth2_file - path to the oauth2 file
+    ##   static_token - YDB static token
+    oauth2_file: "/opt/aardappel/secrets/dst-sa/key.json"
+    ## Overrides the endpoint from the OAuth2 credentials file when set.
+    oauth2_endpoint: "https://tokens.svc.cluster.local:443/oauth2/token/exchange"
 
-## The table name to save replication state in the destination database
-## the table will be created in the destination database if absent.
-## Usually any name which is not clashed with existed tables.
-state_table: "aardappel_state"
-# Instance of replication process. Each aardappel process sharing the same state table should have a unique instance id to avoid mixing states.
-instance_id: "0"
-## Enable standby mode for additional instances with the same instance_id.
-## When enabled, one instance holds the lock and the others wait until it is released.
-multiple_instances_mode: false
-## Enable direct offset commits for each partition.
-## When disabled, Commit is called for each message individually.
-commit_offset_mode: false
-## Maximum number of concurrent offset commit requests when commit_offset_mode is enabled.
-max_concurrent_offset_commits: 1000
+## Reusable topic-to-table mappings; consumers are supplied by each feed.
+stream_sets:
+  - id: replica_tables
+    streams:
+    - src_topic: "producer1/cf1"         # Path of the cdc source topic for the 1st replica table
+      dst_table: "/Root/test/table1_rep" # Path of the 1st replica table (should exist and have compatible schema)
+      problem_strategy: "stop"           # What to do on some problem (stop, continue)
+    - src_topic: "producer2/cf1"         # Path of the cdc source topic for the 2nd replica table
+      dst_table: "/Root/test/table2_rep" # Path of the 2nd replica table (should exist and have compatible schema)
+      problem_strategy: "continue"       # What to do on some problem (stop, continue)
 
-## Replica tables and cdc sources. Transactions will be applied consistently between those tables. Multiple tables can be specified.
-streams:
-  - src_topic: "producer1/cf1"         # Path of the cdc source topic for the 1st replica table
-    consumer: "c1"                     # Topic consumer name in a cdc source topic for the 1st replica table
-    dst_table: "/Root/test/table1_rep" # Path of the 1st replica table (should exist and have compatible schema)
-    problem_strategy: "stop"           # What to do on some problem (stop, continue)
-  - src_topic: "producer2/cf1"         # Path of the cdc source topic for the 2nd replica table
-    consumer: "c1"                     # Topic consumer name in a cdc source topic for the 2nd replica table
-    dst_table: "/Root/test/table2_rep" # Path of the 2nd replica table (should exist and have compatible schema)
-    problem_strategy: "continue"       # What to do on some problem (stop, continue)
+## Replication feeds reference cluster IDs.
+feeds:
+    # Unique feed ID
+  - id: "0"
+    # Optional replication state and lock identifier; defaults to id.
+    # instance_id: "existing-replica"
+    source: source
+    destination: destination
+    ## The table name to save replication state in the destination database
+    ## the table will be created in the destination database if absent.
+    ## Usually any name which is not clashed with existed tables.
+    state_table: "aardappel_state"
+    ## Enable direct offset commits for each partition.
+    ## When disabled, Commit is called for each message individually.
+    commit_offset_mode: false
+    ## Maximum number of concurrent offset commit requests when commit_offset_mode is enabled.
+    max_concurrent_offset_commits: 1000
 
-## Max timeout to analise missed heartbeat in seconds. If we don't get quorum during this interval
-## the warning message will be written to the log.
-## Should be greater than CDC configured heartbeat interval
-max_expected_heartbeat_interval: 10
+    ## Reuse the mappings above with this feed's consumer.
+    streams:
+      - ref: replica_tables
+        consumer: "c1"
 
-## Log level (values "debug", "info", "warn", "error")
-log_level: "debug"
+    ## Max timeout to analyze missed heartbeat in seconds. If we don't get quorum during this interval
+    ## the warning message will be written to the log.
+    ## Should be greater than CDC configured heartbeat interval
+    max_expected_heartbeat_interval: 10
+
+    ## Command topic. In this topic you may post events to suggest aardappel what to do if it stops because cannot process some changes consistently
+    cmd_queue:
+      path: aardappel_command
+      consumer: c1
 
-## Command topic. In this topic you may post events to suggest aardappel what to do if it stops because cannot process some changes consistently
-cmd_queue:
-  path: aardappel_command
-  consumer: c1
+    ## Dead letter queue. In this topic you may get events that haven't processed because they could not be applied consistently
+    dead_letter_queue:
+      path: dlq
 
-## Dead letter queue. In this topic you may get events that haven't processed because they could not be applied consistently
-dead_letter_queue:
-  path: dlq
+## Keep waiting until a feed is available for this instance to own (default true).
+wait_for_feed: true
+## Log level (values "debug", "info", "warn", "error")
+log_level: "debug"
diff --git a/cmd/aardappel/main.go b/cmd/aardappel/main.go
index 4b1a72e..9c361be 100644
--- a/cmd/aardappel/main.go
+++ b/cmd/aardappel/main.go
@@ -116,10 +116,10 @@ func initReplicaStateTable(ctx context.Context, client *client.TableClient, stat
 	return err
 }
 
-func doDescribeTopics(ctx context.Context, config configInit.Config, srcDb *client.TopicClient) hb_tracker.TopicPartsCount {
+func doDescribeTopics(ctx context.Context, feed configInit.Feed, srcDb *client.TopicClient) hb_tracker.TopicPartsCount {
 	topics := hb_tracker.NewTopicPartsCount()
-	for i := 0; i < len(config.Streams); i++ {
-		cfgStream := &config.Streams[i]
+	for i := 0; i < len(feed.Streams); i++ {
+		cfgStream := &feed.Streams[i]
 
 		desc, err := srcDb.Describe(ctx, cfgStream.SrcTopic)
 		if err != nil {
@@ -137,60 +137,60 @@ func doDescribeTopics(ctx context.Context, config configInit.Config, srcDb *clie
 	return *topics
 }
 
-func doMain(ctx context.Context, config configInit.Config, srcDb *client.TopicClient, dstDb *client.YdbClient,
+func doMain(ctx context.Context, feed configInit.Feed, srcDb *client.TopicClient, dstDb *client.YdbClient,
 	locker *ydb_locker.Locker, mon pmon.Metrics, errChannel chan error) error {
-	topics := doDescribeTopics(ctx, config, srcDb)
+	topics := doDescribeTopics(ctx, feed, srcDb)
 
 	xlog.Debug(ctx, "All topics described",
 		zap.Int("total parts", topics.TotalPartsCount))
 
 	var conflictHandler processor.ConflictHandler
 
-	if config.CmdQueue != nil {
-		xlog.Debug(ctx, "Command queue present in config",
-			zap.String("path", config.CmdQueue.Path),
-			zap.String("consumer", config.CmdQueue.Consumer))
+	if feed.CmdQueue != nil {
+		xlog.Debug(ctx, "Command queue present in feed",
+			zap.String("path", feed.CmdQueue.Path),
+			zap.String("consumer", feed.CmdQueue.Consumer))
 		conflictHandler = processor.NewCmdQueueConflictHandler(
-			ctx, config.InstanceId, config.CmdQueue.Path, config.CmdQueue.Consumer, dstDb.TopicClient)
+			ctx, feed.InstanceId, feed.CmdQueue.Path, feed.CmdQueue.Consumer, dstDb.TopicClient)
 	}
 
 	var dlQueue *processor.DLQueue
-	if config.DLQueue != nil {
-		xlog.Info(ctx, "Dead letter queue present in config",
-			zap.String("path", config.DLQueue.Path))
-		topicWriter, err := dstDb.TopicClient.StartWriter(config.DLQueue.Path)
+	if feed.DLQueue != nil {
+		xlog.Info(ctx, "Dead letter queue present in feed",
+			zap.String("path", feed.DLQueue.Path))
+		topicWriter, err := dstDb.TopicClient.StartWriter(feed.DLQueue.Path)
 		if err != nil {
 			xlog.Fatal(ctx, "Unable to start writer for dlq", zap.Error(err))
 		}
 		dlQueue = processor.NewDlQueue(ctx, topicWriter)
 	}
 
-	prc, err := processor.NewProcessor(ctx, topics, config.StateTable, dstDb.TableClient, config.InstanceId, config.KeyFilter, config.CommitOffsetMode, config.MaxConcurrentOffsetCommits)
+	prc, err := processor.NewProcessor(ctx, topics, feed.StateTable, dstDb.TableClient, feed.InstanceId, feed.KeyFilter, feed.CommitOffsetMode, feed.MaxConcurrentOffsetCommits)
 	if err != nil {
 		xlog.Fatal(ctx, "Unable to create processor", zap.Error(err))
 	}
 
-	if config.MaxExpHbInterval != 0 {
-		xlog.Info(ctx, "start heartbeat tracker guard timer", zap.Uint32("timeout in seconds", config.MaxExpHbInterval))
+	if feed.MaxExpHbInterval != 0 {
+		xlog.Info(ctx, "start heartbeat tracker guard timer", zap.Uint32("timeout in seconds", feed.MaxExpHbInterval))
 		if reflect.ValueOf(mon).IsNil() {
-			prc.StartHbGuard(ctx, config.MaxExpHbInterval, nil)
+			prc.StartHbGuard(ctx, feed.MaxExpHbInterval, nil)
 		} else {
-			prc.StartHbGuard(ctx, config.MaxExpHbInterval, mon)
+			prc.StartHbGuard(ctx, feed.MaxExpHbInterval, mon)
 		}
 	}
 
 	var dstTables []*dst_table.DstTable
 
-	for i := 0; i < len(config.Streams); i++ {
-		cfgStream := &config.Streams[i]
+	for i := 0; i < len(feed.Streams); i++ {
+		cfgStream := &feed.Streams[i]
 		var readerOptions []topicoptions.ReaderOption
 		var updateCb topicReader.UpdateOffsetFunc
-		if !config.CommitOffsetMode {
+		if !feed.CommitOffsetMode {
 			var startCb topicoptions.GetPartitionStartOffsetFunc
 			startCb, updateCb = topicReader.MakeTopicReaderGuard(errChannel)
 			readerOptions = append(readerOptions, topicoptions.WithReaderGetPartitionStartOffset(startCb))
 		}
-		reader, err := srcDb.StartReader(config.Streams[i].Consumer, cfgStream.SrcTopic, readerOptions...)
+		reader, err := srcDb.StartReader(cfgStream.Consumer, cfgStream.SrcTopic, readerOptions...)
 
 		if err != nil {
 			xlog.Fatal(ctx, "Unable to create topic reader",
@@ -198,7 +198,7 @@ func doMain(ctx context.Context, config configInit.Config, srcDb *client.TopicCl
 				zap.String("src_topic", cfgStream.SrcTopic),
 				zap.Error(err))
 		}
-		if config.CommitOffsetMode {
+		if feed.CommitOffsetMode {
 			prc.RegisterTopicReader(uint32(i), reader)
 		}
 		dstTables = append(dstTables,
@@ -215,7 +215,7 @@ func doMain(ctx context.Context, config configInit.Config, srcDb *client.TopicCl
 			PartCount:       topics.TopicPartsCountMap[i].PartitionsCount,
 			ProblemStrategy: cfgStream.ProblemStrategy}
 		xlog.Debug(ctx, "Start reading")
-		go topicReader.ReadTopic(ctx, streamInfo, reader, prc, conflictHandler, updateCb, dlQueue, errChannel, config.CommitOffsetMode)
+		go topicReader.ReadTopic(ctx, streamInfo, reader, prc, conflictHandler, updateCb, dlQueue, errChannel, feed.CommitOffsetMode)
 	}
 
 	lockExecutor := func(fn func(context.Context, table.Session, table.Transaction) error) error {
@@ -236,13 +236,13 @@ func doMain(ctx context.Context, config configInit.Config, srcDb *client.TopicCl
 	return nil
 }
 
-func trySrcConnect(ctx context.Context, logger *zap.Logger, config configInit.Config, srcOpts []ydb.Option) bool {
-	srcDb, err := client.NewYdbClient(ctx, logger, config.SrcConnectionString, srcOpts...)
+func trySrcConnect(ctx context.Context, logger *zap.Logger, feed configInit.Feed, cluster configInit.Cluster, srcOpts []ydb.Option) bool {
+	srcDb, err := client.NewYdbClient(ctx, logger, cluster.ConnectionString, srcOpts...)
 	if err != nil {
 		xlog.Fatal(ctx, "Unable to connect to src cluster", zap.Error(err))
 	}
-	res := doDescribeTopics(ctx, config, srcDb.TopicClient)
-	if len(res.TopicPartsCountMap) == len(config.Streams) {
+	res := doDescribeTopics(ctx, feed, srcDb.TopicClient)
+	if len(res.TopicPartsCountMap) == len(feed.Streams) {
 		xlog.Debug(ctx, "Initial check for YDB src connection passed")
 		return true
 	} else {
@@ -250,9 +250,9 @@ func trySrcConnect(ctx context.Context, logger *zap.Logger, config configInit.Co
 	}
 }
 
-func tryDstConnect(ctx context.Context, config configInit.Config, tableClient *client.TableClient) bool {
-	for i := 0; i < len(config.Streams); i++ {
-		cfgStream := &config.Streams[i]
+func tryDstConnect(ctx context.Context, feed configInit.Feed, tableClient *client.TableClient) bool {
+	for i := 0; i < len(feed.Streams); i++ {
+		cfgStream := &feed.Streams[i]
 		_, err := dst_table.DescribeTable(ctx, tableClient, cfgStream.DstTable)
 		if err != nil {
 			return false
@@ -284,6 +284,11 @@ func main() {
 		log.Fatal(ctx, "Unable to initialize config: ", err)
 	}
 
+	// FIXME: Feed selection/locking will be implemented separately.
+	if len(config.Feeds) != 1 {
+		log.Fatal("multiple feeds are not supported yet; configure exactly one feed")
+	}
+
 	// Setup logging
 	logger := xlog.SetupLogging(config.LogLevel)
 	xlog.SetInternalLogger(logger)
@@ -325,16 +330,27 @@ func main() {
 			zap.String("config", confStr))
 	}
 
+	feed := config.Feeds[0]
+
 	var mon *pmon.PromMon
 	if config.MonServer != nil {
-		mon = pmon.NewPromMon(ctx, config.MonServer)
+		mon = pmon.NewPromMon(ctx, config.MonServer, feed.ID)
 		defer mon.Stop()
 	}
 
+	source, err := config.ClusterByID(feed.Source)
+	if err != nil {
+		log.Fatal(err)
+	}
+	destination, err := config.ClusterByID(feed.Destination)
+	if err != nil {
+		log.Fatal(err)
+	}
+
 	srcOpts, err := auth.CreateYdbDriverAuthOptions(auth.AuthConfig{
-		CredentialsFile:   config.SrcOAuthFile,
-		StaticToken:       config.SrcStaticToken,
-		ExchangerEndpoint: config.SrcOAuthEndpoint,
+		CredentialsFile:   source.OAuthFile,
+		StaticToken:       source.StaticToken,
+		ExchangerEndpoint: source.OAuthEndpoint,
 	})
 	if err != nil {
 		xlog.Fatal(ctx, "Unable to create auth option for src",
@@ -342,43 +358,43 @@ func main() {
 	}
 
 	dstOpts, err := auth.CreateYdbDriverAuthOptions(auth.AuthConfig{
-		CredentialsFile:   config.DstOAuthFile,
-		StaticToken:       config.DstStaticToken,
-		ExchangerEndpoint: config.DstOAuthEndpoint,
+		CredentialsFile:   destination.OAuthFile,
+		StaticToken:       destination.StaticToken,
+		ExchangerEndpoint: destination.OAuthEndpoint,
 	})
 	if err != nil {
 		xlog.Fatal(ctx, "Unable to create auth option for dst",
 			zap.Error(err))
 	}
 
-	if config.SrcClientBalancer == false {
+	if source.ClientBalancer == false {
 		srcOpts = append(srcOpts, ydb.WithBalancer(balancers.SingleConn()))
 	}
 
-	if config.DstClientBalancer == false {
+	if destination.ClientBalancer == false {
 		dstOpts = append(dstOpts, ydb.WithBalancer(balancers.SingleConn()))
 	}
 
-	dstDb, err := client.NewYdbClient(ctx, logger, config.DstConnectionString, dstOpts...)
+	dstDb, err := client.NewYdbClient(ctx, logger, destination.ConnectionString, dstOpts...)
 	if err != nil {
 		xlog.Fatal(ctx, "Unable to connect to dst cluster", zap.Error(err))
 	}
 	xlog.Debug(ctx, "YDB dst opened")
 
-	err = createReplicaStateTable(ctx, dstDb.TableClient, config.StateTable)
+	err = createReplicaStateTable(ctx, dstDb.TableClient, feed.StateTable)
 	if err != nil {
 		xlog.Fatal(ctx, "Replication startup failed",
-			zap.String("unable to create table", config.StateTable), zap.Error(err))
+			zap.String("unable to create table", feed.StateTable), zap.Error(err))
 	}
 
-	err = initReplicaStateTable(ctx, dstDb.TableClient, config.StateTable, config.InstanceId)
+	err = initReplicaStateTable(ctx, dstDb.TableClient, feed.StateTable, feed.InstanceId)
 	if err != nil {
 		xlog.Fatal(ctx, "Replication startup failed",
-			zap.String("unable to init table", config.StateTable), zap.Error(err))
+			zap.String("unable to init table", feed.StateTable), zap.Error(err))
 	}
 
-	if trySrcConnect(ctx, logger, config, srcOpts) {
-		if tryDstConnect(ctx, config, dstDb.TableClient) {
+	if trySrcConnect(ctx, logger, feed, source, srcOpts) {
+		if tryDstConnect(ctx, feed, dstDb.TableClient) {
 			if mon != nil {
 				mon.SetCompleted()
 			}
@@ -392,13 +408,13 @@ func main() {
 	hostname, _ := os.Hostname()
 	owner := fmt.Sprintf("lock_%s_%s", hostname, uuid.New().String())
 
-	reqBuilder := GetLockerRequestBuilder(config.StateTable)
+	reqBuilder := GetLockerRequestBuilder(feed.StateTable)
 	lockStorage := ydb_locker.YdbLockStorage{Db: dstDb.GetDriver(), ReqBuilder: reqBuilder}
-	lockTtl := time.Duration(config.MaxExpHbInterval*2) * time.Second
-	locker := ydb_locker.NewLocker(&lockStorage, config.InstanceId, owner, lockTtl)
+	lockTtl := time.Duration(feed.MaxExpHbInterval*2) * time.Second
+	locker := ydb_locker.NewLocker(&lockStorage, feed.InstanceId, owner, lockTtl)
 	lockChannel := locker.LockerContext(ctx)
 	cont := true
-	var lockErrCnt uint32
+	feedWaitChecks := 0
 	for cont {
 		select {
 		case lockCtx, ok := <-lockChannel:
@@ -407,13 +423,12 @@ func main() {
 				cont = false
 				continue
 			}
-			lockErrCnt = 0
-			srcDb, err := client.NewYdbClient(lockCtx, logger, config.SrcConnectionString, srcOpts...)
+			srcDb, err := client.NewYdbClient(lockCtx, logger, source.ConnectionString, srcOpts...)
 			if err != nil {
 				xlog.Fatal(ctx, "Unable to connect to src cluster", zap.Error(err))
 			}
 			xlog.Debug(ctx, "YDB src opened")
-			err = doMain(lockCtx, config, srcDb.TopicClient, dstDb, locker, mon, errChannel)
+			err = doMain(lockCtx, feed, srcDb.TopicClient, dstDb, locker, mon, errChannel)
 			xlog.Info(ctx, "Finished with error, stopping aardappel....", zap.String("error", err.Error()))
 			cancel()
 			cont = false
@@ -426,20 +441,18 @@ func main() {
 				xlog.Fatal(ctx, "Timeout waiting for lock to release")
 			}
 		case <-time.After(defaultLockCheckInterval):
-			if !config.MultipleInstancesMode && lockErrCnt == 10 {
-				cont = false
-				continue
-			}
-			if !config.MultipleInstancesMode {
-				lockErrCnt++
+			if !config.WaitForFeed {
+				feedWaitChecks++
+				if feedWaitChecks >= 10 {
+					xlog.Info(ctx, "No feed ownership acquired, exiting worker")
+					cancel()
+					cont = false
+					continue
+				}
 			}
 			xlog.Info(ctx, "unable to get lock, other instance of aardappel is running")
 		}
 	}
 
-	if lockErrCnt != 0 {
-		xlog.Error(ctx, "aardappel has been shutted down after multiple getting lock errors")
-	} else {
-		xlog.Info(ctx, "aardappel has been shutted down successfully ")
-	}
+	xlog.Info(ctx, "aardappel has been shut down successfully")
 }
diff --git a/internal/auth/README.md b/internal/auth/README.md
index 61f086b..7e8c44e 100644
--- a/internal/auth/README.md
+++ b/internal/auth/README.md
@@ -4,8 +4,8 @@ This package creates YDB driver authentication options for aardappel.
 
 At the aardappel config level, each side of replication can use exactly one of:
 
-- `*_static_token`: a final YDB access token, used directly without OAuth2 token exchange.
-- `*_oauth2_file`: a credentials file used to configure OAuth2 token exchange.
+- `static_token`: a final YDB access token, used directly without OAuth2 token exchange.
+- `oauth2_file`: a credentials file used to configure OAuth2 token exchange.
 
 For OAuth2 credentials, aardappel currently supports two JSON file formats.
 
@@ -94,10 +94,10 @@ The optional exchanger endpoint is configured as:
 
 ## Endpoint Override
 
-The aardappel config can specify side-specific OAuth2 exchanger endpoints:
+Each cluster can specify its OAuth2 exchanger endpoint with `oauth2_endpoint`.
 
-- `src_oauth2_endpoint`
-- `dst_oauth2_endpoint`
+- `oauth2_endpoint`
+- `oauth2_endpoint`
 
 When a side-specific endpoint is set in aardappel config, it overrides the endpoint from the credentials file.
 
diff --git a/internal/config/config.go b/internal/config/config.go
index f60f9cf..4206ac3 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -6,10 +6,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"go.uber.org/zap"
-	"gopkg.in/yaml.v3"
 	"os"
+	"slices"
 	"strings"
+
+	"go.uber.org/zap"
+	"gopkg.in/yaml.v3"
 )
 
 type Stream struct {
@@ -20,6 +22,16 @@ type Stream struct {
 	MonTag          string `yaml:"mon_tag"`
 }
 
+type StreamSet struct {
+	ID      string   `yaml:"id"`
+	Streams []Stream `yaml:"streams"`
+}
+
+type StreamReference struct {
+	Ref      string `yaml:"ref"`
+	Consumer string `yaml:"consumer"`
+}
+
 type MonServer struct {
 	Listen string `yaml:"listen"`
 }
@@ -38,31 +50,49 @@ type KeyFilter struct {
 }
 
 type Config struct {
-	SrcConnectionString   string     `yaml:"src_connection_string"`
-	SrcClientBalancer     bool       `yaml:"src_client_balancer"`
-	SrcOAuthFile          string     `yaml:"src_oauth2_file"`
-	SrcOAuthEndpoint      string     `yaml:"src_oauth2_endpoint"`
-	SrcStaticToken        string     `yaml:"src_static_token"`
-	DstConnectionString   string     `yaml:"dst_connection_string"`
-	DstClientBalancer     bool       `yaml:"dst_client_balancer"`
-	DstOAuthFile          string     `yaml:"dst_oauth2_file"`
-	DstOAuthEndpoint      string     `yaml:"dst_oauth2_endpoint"`
-	DstStaticToken        string     `yaml:"dst_static_token"`
-	InstanceId            string     `yaml:"instance_id"`
-	MultipleInstancesMode bool       `yaml:"multiple_instances_mode"`
-	CommitOffsetMode      bool       `yaml:"commit_offset_mode"`
-	Streams               []Stream   `yaml:"streams"`
-	StateTable            string     `yaml:"state_table"`
-	MaxExpHbInterval      uint32     `yaml:"max_expected_heartbeat_interval"`
-	LogLevel              string     `yaml:"log_level"`
-	MonServer             *MonServer `yaml:"mon_server"`
-	CmdQueue              *CmdQueue  `yaml:"cmd_queue"`
-	KeyFilter             *KeyFilter `yaml:"key_filter"`
-	DLQueue               *DLQueue   `yaml:"dead_letter_queue"`
+	Clusters    []Cluster   `yaml:"clusters"`
+	StreamSets  []StreamSet `yaml:"stream_sets,omitempty"`
+	Feeds       []Feed      `yaml:"feeds"`
+	WaitForFeed bool        `yaml:"wait_for_feed"`
+	LogLevel    string      `yaml:"log_level"`
+	MonServer   *MonServer  `yaml:"mon_server"`
+}
+
+type Cluster struct {
+	ID               string            `yaml:"id"`
+	ConnectionString string            `yaml:"connection_string"`
+	ClientBalancer   bool              `yaml:"client_balancer"`
+	OAuthFile        string            `yaml:"oauth2_file"`
+	OAuthEndpoint    string            `yaml:"oauth2_endpoint"`
+	StaticToken      string            `yaml:"static_token"`
+}
+
+type Feed struct {
+	InstanceId       string            `yaml:"instance_id,omitempty"`
+	Source           string            `yaml:"source"`
+	Destination      string            `yaml:"destination"`
+	ID               string            `yaml:"id"`
+	CommitOffsetMode bool              `yaml:"commit_offset_mode"`
+	Streams          []Stream          `yaml:"streams"`
+	StreamReferences []StreamReference `yaml:"-"`
+	StateTable       string            `yaml:"state_table"`
+	MaxExpHbInterval uint32            `yaml:"max_expected_heartbeat_interval"`
+	CmdQueue         *CmdQueue         `yaml:"cmd_queue"`
+	KeyFilter        *KeyFilter        `yaml:"key_filter"`
+	DLQueue          *DLQueue          `yaml:"dead_letter_queue"`
 
 	MaxConcurrentOffsetCommits int `yaml:"max_concurrent_offset_commits"`
 }
 
+func (config *Config) ClusterByID(id string) (Cluster, error) {
+	for _, cluster := range config.Clusters {
+		if cluster.ID == id {
+			return cluster, nil
+		}
+	}
+	return Cluster{}, fmt.Errorf("unknown cluster id %q", id)
+}
+
 func verifyStreamProblemStrategy(configStrategy *string) error {
 	if *configStrategy == "" {
 		*configStrategy = types.ProblemStrategyStop
@@ -70,7 +100,7 @@ func verifyStreamProblemStrategy(configStrategy *string) error {
 	}
 	streamProblemStrategies := []string{types.ProblemStrategyStop, types.ProblemStrategyContinue}
 	for _, strategy := range streamProblemStrategies {
-		if strings.ToLower(strategy) == strings.ToLower(*configStrategy) {
+		if strings.EqualFold(strategy, *configStrategy) {
 			*configStrategy = strategy
 			return nil
 		}
@@ -78,20 +108,232 @@ func verifyStreamProblemStrategy(configStrategy *string) error {
 	return errors.New("unknown stream problem strategy '" + *configStrategy + "'")
 }
 
-func (config Config) ToString() (string, error) {
-	data, err := yaml.Marshal(&config)
+func (config *Config) ToString() (string, error) {
+	data, err := yaml.Marshal(config)
 	if err != nil {
 		return "", err
 	}
-	return string(data), nil
+	return "version: 2\n" + string(data), nil
+}
+
+func (config *Config) verify() error {
+	if len(config.Feeds) == 0 {
+		return errors.New("at least one feed is required")
+	}
+	if len(config.Clusters) == 0 {
+		return errors.New("at least one cluster is required")
+	}
+	clusterIDs := make(map[string]struct{})
+	for i, cluster := range config.Clusters {
+		clusterID := cluster.ID
+		if strings.TrimSpace(clusterID) == "" {
+			return fmt.Errorf("clusters[%d]: id is required", i)
+		}
+		if _, exists := clusterIDs[clusterID]; exists {
+			return fmt.Errorf("clusters[%d]: duplicate cluster id %q", i, clusterID)
+		}
+		clusterIDs[clusterID] = struct{}{}
+		if strings.TrimSpace(cluster.ConnectionString) == "" {
+			return fmt.Errorf("clusters[%d]: connection_string is required", i)
+		}
+	}
+	streamSets := make(map[string][]Stream)
+	for i := range config.StreamSets {
+		set := &config.StreamSets[i]
+		if strings.TrimSpace(set.ID) == "" {
+			return fmt.Errorf("stream_sets[%d]: id is required", i)
+		}
+		if _, exists := streamSets[set.ID]; exists {
+			return fmt.Errorf("stream_sets[%d]: duplicate stream set id %q", i, set.ID)
+		}
+		if len(set.Streams) == 0 {
+			return fmt.Errorf("stream_sets[%d]: at least one stream is required", i)
+		}
+		for j := range set.Streams {
+			if set.Streams[j].Consumer != "" {
+				return fmt.Errorf("stream_sets[%d].streams[%d]: consumer belongs to the referencing feed", i, j)
+			}
+			if err := verifyStreamProblemStrategy(&set.Streams[j].ProblemStrategy); err != nil {
+				return fmt.Errorf("stream_sets[%d].streams[%d]: %w", i, j, err)
+			}
+		}
+		streamSets[set.ID] = set.Streams
+	}
+	feedIDs := make(map[string]struct{})
+	for i := range config.Feeds {
+		feed := &config.Feeds[i]
+		if strings.TrimSpace(feed.ID) == "" {
+			return fmt.Errorf("feeds[%d]: id is required", i)
+		}
+		if _, exists := feedIDs[feed.ID]; exists {
+			return fmt.Errorf("feeds[%d]: duplicate feed id %q", i, feed.ID)
+		}
+		feedIDs[feed.ID] = struct{}{}
+		if feed.InstanceId == "" {
+			feed.InstanceId = feed.ID
+		}
+		refs := feed.StreamReferences
+		if len(refs) > 0 {
+			feed.Streams = nil
+			for j, ref := range refs {
+				streams, exists := streamSets[ref.Ref]
+				if !exists {
+					return fmt.Errorf("feeds[%d].streams[%d]: unknown stream set id %q", i, j, ref.Ref)
+				}
+				for _, stream := range streams {
+					stream.Consumer = ref.Consumer
+					feed.Streams = append(feed.Streams, stream)
+				}
+			}
+		}
+		if err := feed.verify(clusterIDs); err != nil {
+			return fmt.Errorf("feeds[%d]: %w", i, err)
+		}
+	}
+	return nil
+}
+
+func (config *Config) warnSharedInstances(ctx context.Context) {
+	type instanceDestination struct {
+		instanceID  string
+		destination string
+	}
+	seen := make(map[instanceDestination]string)
+	for _, feed := range config.Feeds {
+		key := instanceDestination{feed.InstanceId, feed.Destination}
+		if firstFeed, exists := seen[key]; exists {
+			xlog.Warn(ctx, "Different feeds use the same instance_id and destination; check that this is intended",
+				zap.String("feed_id", feed.ID),
+				zap.String("other_feed_id", firstFeed),
+				zap.String("instance_id", feed.InstanceId),
+				zap.String("destination", feed.Destination))
+		} else {
+			seen[key] = feed.ID
+		}
+	}
+}
+
+func (feed *Feed) UnmarshalYAML(value *yaml.Node) error {
+	type plain Feed
+	if value.Kind != yaml.MappingNode {
+		return errors.New("feed must be a mapping")
+	}
+	defaults := plain{MaxConcurrentOffsetCommits: 1000}
+	var wire struct{ Streams yaml.Node }
+	rest := *value
+	rest.Content = nil
+	for i := 0; i < len(value.Content); i += 2 {
+		if value.Content[i].Value == "streams" {
+			if wire.Streams.Kind != 0 {
+				return errors.New("duplicate streams field")
+			}
+			wire.Streams = *value.Content[i+1]
+			for wire.Streams.Kind == yaml.AliasNode {
+				wire.Streams = *wire.Streams.Alias
+			}
+		} else {
+			rest.Content = append(rest.Content, value.Content[i], value.Content[i+1])
+		}
+	}
+	if err := rest.Decode(&defaults); err != nil {
+		return err
+	}
+	switch wire.Streams.Kind {
+	case yaml.SequenceNode:
+		hasReferences := slices.ContainsFunc(wire.Streams.Content, isStreamReference)
+		if hasReferences {
+			for i, entry := range wire.Streams.Content {
+				ref, err := decodeStreamReference(entry)
+				if err != nil {
+					return fmt.Errorf("streams[%d]: %w", i, err)
+				}
+				defaults.StreamReferences = append(defaults.StreamReferences, ref)
+			}
+		} else if err := wire.Streams.Decode(&defaults.Streams); err != nil {
+			return err
+		}
+	case 0, yaml.ScalarNode:
+		if wire.Streams.Kind != 0 && wire.Streams.Tag != "!!null" {
+			return errors.New("streams must be a list of inline streams or references")
+		}
+	default:
+		return errors.New("streams must be a list of inline streams or references")
+	}
+	*feed = Feed(defaults)
+	return nil
 }
 
-func (config Config) verify() error {
-	if config.CommitOffsetMode && config.MaxConcurrentOffsetCommits <= 0 {
+func isStreamReference(node *yaml.Node) bool {
+	for node.Kind == yaml.AliasNode {
+		node = node.Alias
+	}
+	if node.Kind != yaml.MappingNode {
+		return false
+	}
+	for i := 0; i < len(node.Content); i += 2 {
+		if node.Content[i].Value == "ref" {
+			return true
+		}
+	}
+	return false
+}
+
+func decodeStreamReference(node *yaml.Node) (StreamReference, error) {
+	for node.Kind == yaml.AliasNode {
+		node = node.Alias
+	}
+	var ref StreamReference
+	if node.Kind != yaml.MappingNode {
+		return ref, errors.New("streams reference must be a mapping")
+	}
+	for i := 0; i < len(node.Content); i += 2 {
+		key := node.Content[i].Value
+		if key != "ref" && key != "consumer" {
+			return ref, fmt.Errorf("streams reference: unknown field %q", key)
+		}
+	}
+	if err := node.Decode(&ref); err != nil {
+		return ref, err
+	}
+	if strings.TrimSpace(ref.Ref) == "" {
+		return ref, errors.New("streams reference: ref is required")
+	}
+	if strings.TrimSpace(ref.Consumer) == "" {
+		return ref, errors.New("streams reference: consumer is required")
+	}
+	return ref, nil
+}
+
+func (feed Feed) MarshalYAML() (any, error) {
+	type plain Feed
+	var node yaml.Node
+	if err := node.Encode(plain(feed)); err != nil {
+		return nil, err
+	}
+	if len(feed.StreamReferences) > 0 {
+		for i := 0; i < len(node.Content); i += 2 {
+			if node.Content[i].Value == "streams" {
+				if err := node.Content[i+1].Encode(feed.StreamReferences); err != nil {
+					return nil, err
+				}
+			}
+		}
+	}
+	return &node, nil
+}
+
+func (feed *Feed) verify(knownIDs map[string]struct{}) error {
+	if _, known := knownIDs[feed.Source]; !known {
+		return fmt.Errorf("unknown source cluster id %q", feed.Source)
+	}
+	if _, known := knownIDs[feed.Destination]; !known {
+		return fmt.Errorf("unknown destination cluster id %q", feed.Destination)
+	}
+	if feed.CommitOffsetMode && feed.MaxConcurrentOffsetCommits <= 0 {
 		return errors.New("max_concurrent_offset_commits must be greater than zero when commit_offset_mode is enabled")
 	}
-	for _, stream := range config.Streams {
-		err := verifyStreamProblemStrategy(&stream.ProblemStrategy)
+	for i := range feed.Streams {
+		err := verifyStreamProblemStrategy(&feed.Streams[i].ProblemStrategy)
 		if err != nil {
 			return err
 		}
@@ -110,8 +352,21 @@ func InitConfig(ctx context.Context, confPath string) (Config, error) {
 			zap.Error(err))
 		return Config{}, fmt.Errorf("unable to read configuration file: %w", err)
 	}
-	config := Config{MaxConcurrentOffsetCommits: 1000}
-	err = yaml.Unmarshal(confTxt, &config)
+	// Only a version marker on the first line selects the new format.
+	firstLine, _, _ := strings.Cut(string(confTxt), "\n")
+	var header struct {
+		Version int `yaml:"version"`
+	}
+	config := Config{WaitForFeed: true}
+	if yaml.Unmarshal([]byte(firstLine), &header) == nil && header.Version == 2 {
+		err = yaml.Unmarshal(confTxt, &config)
+	} else {
+		xlog.Warn(ctx, "Legacy configuration format is in use; migrate to the new format with version: 2 on the first line",
+			zap.String("config_path", confPath))
+		legacy := legacyConfig{MaxConcurrentOffsetCommits: 1000}
+		err = yaml.Unmarshal(confTxt, &legacy)
+		config = legacy.toConfig()
+	}
 	if err != nil {
 		xlog.Error(ctx, "Unable to parse configuration file",
 			zap.String("config_path", confPath),
@@ -123,5 +378,6 @@ func InitConfig(ctx context.Context, confPath string) (Config, error) {
 		xlog.Error(ctx, "Unable to verify configuration file", zap.Error(err))
 		return Config{}, fmt.Errorf("unable to verify configuration file: %w", err)
 	}
+	config.warnSharedInstances(ctx)
 	return config, nil
 }
diff --git a/internal/config/legacy.go b/internal/config/legacy.go
new file mode 100644
index 0000000..3365ba4
--- /dev/null
+++ b/internal/config/legacy.go
@@ -0,0 +1,45 @@
+package config
+
+type legacyConfig struct {
+	SrcConnectionString   string     `yaml:"src_connection_string"`
+	SrcClientBalancer     bool       `yaml:"src_client_balancer"`
+	SrcOAuthFile          string     `yaml:"src_oauth2_file"`
+	SrcOAuthEndpoint      string     `yaml:"src_oauth2_endpoint"`
+	SrcStaticToken        string     `yaml:"src_static_token"`
+	DstConnectionString   string     `yaml:"dst_connection_string"`
+	DstClientBalancer     bool       `yaml:"dst_client_balancer"`
+	DstOAuthFile          string     `yaml:"dst_oauth2_file"`
+	DstOAuthEndpoint      string     `yaml:"dst_oauth2_endpoint"`
+	DstStaticToken        string     `yaml:"dst_static_token"`
+	InstanceId            string     `yaml:"instance_id"`
+	MultipleInstancesMode bool       `yaml:"multiple_instances_mode"`
+	CommitOffsetMode      bool       `yaml:"commit_offset_mode"`
+	Streams               []Stream   `yaml:"streams"`
+	StateTable            string     `yaml:"state_table"`
+	MaxExpHbInterval      uint32     `yaml:"max_expected_heartbeat_interval"`
+	LogLevel              string     `yaml:"log_level"`
+	MonServer             *MonServer `yaml:"mon_server"`
+	CmdQueue              *CmdQueue  `yaml:"cmd_queue"`
+	KeyFilter             *KeyFilter `yaml:"key_filter"`
+	DLQueue               *DLQueue   `yaml:"dead_letter_queue"`
+
+	MaxConcurrentOffsetCommits int `yaml:"max_concurrent_offset_commits"`
+}
+
+func (legacy legacyConfig) toConfig() Config {
+	current := Config{LogLevel: legacy.LogLevel, MonServer: legacy.MonServer}
+	current.Clusters = []Cluster{
+		{ID: "source", ConnectionString: legacy.SrcConnectionString, ClientBalancer: legacy.SrcClientBalancer,
+			OAuthFile: legacy.SrcOAuthFile, OAuthEndpoint: legacy.SrcOAuthEndpoint, StaticToken: legacy.SrcStaticToken},
+		{ID: "destination", ConnectionString: legacy.DstConnectionString, ClientBalancer: legacy.DstClientBalancer,
+			OAuthFile: legacy.DstOAuthFile, OAuthEndpoint: legacy.DstOAuthEndpoint, StaticToken: legacy.DstStaticToken},
+	}
+	current.Feeds = []Feed{{
+		ID: legacy.InstanceId, Source: "source", Destination: "destination",
+		CommitOffsetMode: legacy.CommitOffsetMode, MaxConcurrentOffsetCommits: legacy.MaxConcurrentOffsetCommits,
+		Streams: legacy.Streams, StateTable: legacy.StateTable, MaxExpHbInterval: legacy.MaxExpHbInterval,
+		CmdQueue: legacy.CmdQueue, KeyFilter: legacy.KeyFilter, DLQueue: legacy.DLQueue,
+	}}
+	current.WaitForFeed = legacy.MultipleInstancesMode
+	return current
+}
diff --git a/internal/config/legacy_test.go b/internal/config/legacy_test.go
new file mode 100644
index 0000000..387fee0
--- /dev/null
+++ b/internal/config/legacy_test.go
@@ -0,0 +1,81 @@
+package config
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"aardappel/internal/types"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestLegacyConfigTransition(t *testing.T) {
+	const legacyYAML = `src_connection_string: grpc://source:2136/source
+dst_connection_string: grpc://destination:2136/destination
+instance_id: replication
+streams:
+  - src_topic: orders/updates
+    dst_table: orders
+    consumer: replication-consumer
+    problem_strategy: continue
+    mon_tag: orders-monitor
+src_client_balancer: true
+src_oauth2_file: source-oauth.json
+src_oauth2_endpoint: https://source.example/token
+src_static_token: source-token
+dst_client_balancer: true
+dst_oauth2_file: destination-oauth.json
+dst_oauth2_endpoint: https://destination.example/token
+dst_static_token: destination-token
+multiple_instances_mode: true
+commit_offset_mode: true
+max_concurrent_offset_commits: 42
+state_table: replication/state
+max_expected_heartbeat_interval: 30
+log_level: debug
+mon_server:
+  listen: localhost:8080
+cmd_queue:
+  path: replication/commands
+  consumer: command-consumer
+key_filter:
+  table_path: replication/keys
+dead_letter_queue:
+  path: replication/dead-letters
+`
+	expected := Config{
+		Clusters: []Cluster{
+			{
+				ID: "source", ConnectionString: "grpc://source:2136/source", ClientBalancer: true,
+				OAuthFile: "source-oauth.json", OAuthEndpoint: "https://source.example/token", StaticToken: "source-token",
+			},
+			{
+				ID: "destination", ConnectionString: "grpc://destination:2136/destination", ClientBalancer: true,
+				OAuthFile: "destination-oauth.json", OAuthEndpoint: "https://destination.example/token", StaticToken: "destination-token",
+			},
+		},
+		WaitForFeed: true,
+		LogLevel:    "debug",
+		MonServer:   &MonServer{Listen: "localhost:8080"},
+		Feeds: []Feed{{
+			ID: "replication", InstanceId: "replication", Source: "source", Destination: "destination",
+			CommitOffsetMode: true, MaxConcurrentOffsetCommits: 42,
+			Streams: []Stream{{
+				SrcTopic: "orders/updates", DstTable: "orders", Consumer: "replication-consumer",
+				ProblemStrategy: types.ProblemStrategyContinue, MonTag: "orders-monitor",
+			}},
+			StateTable: "replication/state", MaxExpHbInterval: 30,
+			CmdQueue:  &CmdQueue{Path: "replication/commands", Consumer: "command-consumer"},
+			KeyFilter: &KeyFilter{Path: "replication/keys"},
+			DLQueue:   &DLQueue{Path: "replication/dead-letters"},
+		}},
+	}
+
+	path := filepath.Join(t.TempDir(), "config.yaml")
+	require.NoError(t, os.WriteFile(path, []byte(legacyYAML), 0600))
+	converted, err := InitConfig(context.Background(), path)
+	require.NoError(t, err)
+	require.Equal(t, expected, converted)
+}
diff --git a/internal/pmon/pmon.go b/internal/pmon/pmon.go
index 2f77af8..f2aa846 100644
--- a/internal/pmon/pmon.go
+++ b/internal/pmon/pmon.go
@@ -7,8 +7,8 @@ import (
 	"context"
 	"errors"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/prometheus/client_golang/prometheus/collectors"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"go.uber.org/zap"
 	"net/http"
 	"sync"
@@ -36,18 +36,18 @@ type MonPerTable struct {
 	topicWithoutHb            prometheus.Gauge
 }
 
-func CreatePerTableCounters(tag string, reg *prometheus.Registry) *MonPerTable {
+func CreatePerTableCounters(tag string, reg *prometheus.Registry, feedID string) *MonPerTable {
 	m := &MonPerTable{
 		modificationsCountByTopic: prometheus.NewCounter(prometheus.CounterOpts{
 			Name:        "modifications_count_per_table",
 			Help:        "Total count of modifications per table on the destination cluster.",
-			ConstLabels: prometheus.Labels{"stream_tag": tag},
+			ConstLabels: prometheus.Labels{"feed_id": feedID, "stream_tag": tag},
 		}),
 
 		topicWithoutHb: prometheus.NewGauge(prometheus.GaugeOpts{
 			Name:        "topic_without_hb",
 			Help:        "topic where hb was not recieved during interval",
-			ConstLabels: prometheus.Labels{"stream_tag": tag},
+			ConstLabels: prometheus.Labels{"feed_id": feedID, "stream_tag": tag},
 		}),
 	}
 
@@ -59,6 +59,7 @@ func CreatePerTableCounters(tag string, reg *prometheus.Registry) *MonPerTable {
 
 type PromMon struct {
 	reg                  *prometheus.Registry
+	feedID               string
 	modificationsCount   prometheus.Counter
 	commitLatency        prometheus.Histogram
 	requestSize          prometheus.Counter
@@ -80,39 +81,46 @@ func lazyMetrics(mon *PromMon, tag string) *MonPerTable {
 	if ok {
 		return counter
 	} else {
-		tmp := CreatePerTableCounters(tag, mon.reg)
+		tmp := CreatePerTableCounters(tag, mon.reg, mon.feedID)
 		mon.perTableCounters[tag] = tmp
 		return tmp
 	}
 }
 
-func NewMetrics(reg *prometheus.Registry) *PromMon {
+func NewMetrics(reg *prometheus.Registry, feedID string) *PromMon {
 	m := &PromMon{
+		feedID: feedID,
 		modificationsCount: prometheus.NewCounter(prometheus.CounterOpts{
-			Name: "modifications_count",
-			Help: "Total count of modifications on the destination cluster.",
+			ConstLabels: prometheus.Labels{"feed_id": feedID},
+			Name:        "modifications_count",
+			Help:        "Total count of modifications on the destination cluster.",
 		}),
 		commitLatency: prometheus.NewHistogram(prometheus.HistogramOpts{
-			Name:    "commit_latency",
-			Help:    "Latency of commit changes on the destination cluster (seconds).",
-			Buckets: prometheus.DefBuckets,
+			ConstLabels: prometheus.Labels{"feed_id": feedID},
+			Name:        "commit_latency",
+			Help:        "Latency of commit changes on the destination cluster (seconds).",
+			Buckets:     prometheus.DefBuckets,
 		}),
 		requestSize: prometheus.NewCounter(prometheus.CounterOpts{
-			Name: "request_size_bytes",
-			Help: "Size of request on the destination cluster (bytes).",
+			ConstLabels: prometheus.Labels{"feed_id": feedID},
+			Name:        "request_size_bytes",
+			Help:        "Size of request on the destination cluster (bytes).",
 		}),
 		quorumWaitingLatency: prometheus.NewHistogram(prometheus.HistogramOpts{
-			Name:    "quorum_waiting_latency",
-			Help:    "Latency of waiting quorum changes on the destination cluster (seconds).",
-			Buckets: prometheus.DefBuckets,
+			ConstLabels: prometheus.Labels{"feed_id": feedID},
+			Name:        "quorum_waiting_latency",
+			Help:        "Latency of waiting quorum changes on the destination cluster (seconds).",
+			Buckets:     prometheus.DefBuckets,
 		}),
 		heapAllocated: prometheus.NewGauge(prometheus.GaugeOpts{
-			Name: "go_heap_allocated",
-			Help: "Size is bytes of allocated heap objects",
+			ConstLabels: prometheus.Labels{"feed_id": feedID},
+			Name:        "go_heap_allocated",
+			Help:        "Size is bytes of allocated heap objects",
 		}),
 		replicationLagEst: prometheus.NewGauge(prometheus.GaugeOpts{
-			Name: "replication_lag_estimation",
-			Help: "The time difference between coordination time of quorum hb and local time",
+			ConstLabels: prometheus.Labels{"feed_id": feedID},
+			Name:        "replication_lag_estimation",
+			Help:        "The time difference between coordination time of quorum hb and local time",
 		}),
 	}
 	reg.MustRegister(m.modificationsCount)
@@ -140,7 +148,7 @@ func (h *khzHandler) ServeHTTP(writer http.ResponseWriter, req *http.Request) {
 	}
 }
 
-func NewPromMon(ctx context.Context, config *config.MonServer) *PromMon {
+func NewPromMon(ctx context.Context, config *config.MonServer, feedID string) *PromMon {
 	reg := prometheus.NewRegistry()
 
 	reg.MustRegister(
@@ -151,7 +159,7 @@ func NewPromMon(ctx context.Context, config *config.MonServer) *PromMon {
 		),
 	)
 
-	p := NewMetrics(reg)
+	p := NewMetrics(reg, feedID)
 	p.perTableCounters = make(map[string]*MonPerTable)
 	p.reg = reg
 
diff --git a/internal/pmon/pmon_test.go b/internal/pmon/pmon_test.go
index ec42e26..afa7ed2 100644
--- a/internal/pmon/pmon_test.go
+++ b/internal/pmon/pmon_test.go
@@ -6,13 +6,14 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"github.com/stretchr/testify/assert"
 	"io"
 	"net/http"
 	"os"
 	"strconv"
 	"testing"
 	"time"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func findField(resBody []byte, pattern []byte) int {
@@ -38,7 +39,7 @@ func findField(resBody []byte, pattern []byte) int {
 
 func TestMonCount(t *testing.T) {
 	ctx := context.Background()
-	p := pmon.NewPromMon(ctx, &config.MonServer{":8080"})
+	p := pmon.NewPromMon(ctx, &config.MonServer{":8080"}, "test-feed")
 	defer p.Stop()
 	p.ModificationCount(123)
 	p.ModificationCountFromTopic(456, "tag_1")
@@ -72,16 +73,16 @@ func TestMonCount(t *testing.T) {
 	assert.NoError(t, err)
 	fmt.Fprintf(os.Stderr, "%v\n", string(resBody))
 
-	pattern_mc := []byte("modifications_count")
+	pattern_mc := []byte(`modifications_count{feed_id="test-feed"}`)
 	val_mc := findField(resBody, pattern_mc)
 
-	pattern_mc_per_table := []byte("modifications_count_per_table{stream_tag=\"tag_1\"}")
+	pattern_mc_per_table := []byte("modifications_count_per_table{feed_id=\"test-feed\",stream_tag=\"tag_1\"}")
 	val_mc_per_table := findField(resBody, pattern_mc_per_table)
 
-	patterm_nh_per_table_1 := []byte("topic_without_hb{stream_tag=\"tag_1\"}")
+	patterm_nh_per_table_1 := []byte("topic_without_hb{feed_id=\"test-feed\",stream_tag=\"tag_1\"}")
 	val_nh_per_table_1 := findField(resBody, patterm_nh_per_table_1)
 
-	patterm_nh_per_table_2 := []byte("topic_without_hb{stream_tag=\"tag_2\"}")
+	patterm_nh_per_table_2 := []byte("topic_without_hb{feed_id=\"test-feed\",stream_tag=\"tag_2\"}")
 	val_nh_per_table_2 := findField(resBody, patterm_nh_per_table_2)
 
 	assert.Equal(t, val_mc, 123)