```
2020.09.24 13:54:06.581506 [ 17189 ] {} <Debug> StorageDistributed (.read_shard_0.destination_cluster.dwh._dim_customer_local): Auto-increment is 0
2020.09.24 13:54:06.581537 [ 17192 ] {} <Debug> StorageDistributed (.read_shard_3.destination_cluster.dwh._dim_customer_local): Auto-increment is 0
2020.09.24 13:54:06.581681 [ 17189 ] {} <Debug> ClusterCopier: Computing destination partition set, executing query: SELECT DISTINCT 'all' AS partition FROM _local.`.read_shard_0.destination_cluster.dwh._dim_customer_local` ORDER BY partition DESC
2020.09.24 13:54:06.581804 [ 17192 ] {} <Debug> ClusterCopier: Computing destination partition set, executing query: SELECT DISTINCT 'all' AS partition FROM _local.`.read_shard_3.destination_cluster.dwh._dim_customer_local` ORDER BY partition DESC
2020.09.24 13:54:06.581929 [ 17191 ] {} <Debug> StorageDistributed (.read_shard_4.destination_cluster.dwh._dim_customer_local): Auto-increment is 0
2020.09.24 13:54:06.582132 [ 17191 ] {} <Debug> ClusterCopier: Computing destination partition set, executing query: SELECT DISTINCT 'all' AS partition FROM _local.`.read_shard_4.destination_cluster.dwh._dim_customer_local` ORDER BY partition DESC
2020.09.24 13:54:06.582538 [ 17189 ] {} <Trace> InterpreterSelectQuery: Complete -> Complete
2020.09.24 13:54:06.582997 [ 17191 ] {} <Trace> InterpreterSelectQuery: Complete -> Complete
2020.09.24 13:54:06.583306 [ 17190 ] {} <Debug> StorageDistributed (.read_shard_1.destination_cluster.dwh._dim_customer_local): Auto-increment is 0
2020.09.24 13:54:06.583470 [ 17190 ] {} <Debug> ClusterCopier: Computing destination partition set, executing query: SELECT DISTINCT 'all' AS partition FROM _local.`.read_shard_1.destination_cluster.dwh._dim_customer_local` ORDER BY partition DESC
2020.09.24 13:54:06.583989 [ 17192 ] {} <Trace> InterpreterSelectQuery: Complete -> Complete
2020.09.24 13:54:06.584162 [ 17190 ] {} <Trace> InterpreterSelectQuery: Complete -> Complete
2020.09.24 13:54:06.585082 [ 17188 ] {} <Debug> StorageDistributed (.read_shard_2.destination_cluster.dwh._dim_customer_local): Auto-increment is 0
2020.09.24 13:54:06.585215 [ 17188 ] {} <Debug> ClusterCopier: Computing destination partition set, executing query: SELECT DISTINCT 'all' AS partition FROM _local.`.read_shard_2.destination_cluster.dwh._dim_customer_local` ORDER BY partition DESC
2020.09.24 13:54:06.585858 [ 17188 ] {} <Trace> InterpreterSelectQuery: Complete -> Complete
2020.09.24 13:54:06.587341 [ 17171 ] {} <Trace> BaseDaemon: Received signal 11



2020.09.24 13:54:06.587515 [ 17193 ] {} <Fatal> BaseDaemon: ########################################
2020.09.24 13:54:06.587579 [ 17193 ] {} <Fatal> BaseDaemon: (version 20.6.6.7, no build id) (from thread 17191) (no query) Received signal Segmentation fault (11)
2020.09.24 13:54:06.587693 [ 17193 ] {} <Fatal> BaseDaemon: Address: 0x7f5b55deff4f Access: read. Address not mapped to object.
2020.09.24 13:54:06.587710 [ 17193 ] {} <Fatal> BaseDaemon: Stack trace: 0xd2380e1 0x5d048ff 0xa2a53ef 0x5dfc737 0x5dfd2c7 0x5d0d06d 0x5d0d753 0x5d0c60d 0x5d0acdf 0x7f5b07ba5e65 0x7f5b074c288d
2020.09.24 13:54:06.587767 [ 17193 ] {} <Fatal> BaseDaemon: 3. memcpy @ 0xd2380e1 in /usr/bin/clickhouse
2020.09.24 13:54:06.587836 [ 17193 ] {} <Fatal> BaseDaemon: 4. void DB::writeAnyEscapedString<(char)39, false>(char const*, char const*, DB::WriteBuffer&) @ 0x5d048ff in /usr/bin/clickhouse
2020.09.24 13:54:06.587870 [ 17193 ] {} <Fatal> BaseDaemon: 5. DB::DataTypeString::serializeTextQuoted(DB::IColumn const&, unsigned long, DB::WriteBuffer&, DB::FormatSettings const&) const @ 0xa2a53ef in /usr/bin/clickhouse
2020.09.24 13:54:06.587890 [ 17193 ] {} <Fatal> BaseDaemon: 6. DB::ClusterCopier::getShardPartitions(DB::ConnectionTimeouts const&, DB::TaskShard&) @ 0x5dfc737 in /usr/bin/clickhouse
2020.09.24 13:54:06.587933 [ 17193 ] {} <Fatal> BaseDaemon: 7. DB::ClusterCopier::discoverShardPartitions(DB::ConnectionTimeouts const&, std::__1::shared_ptr<DB::TaskShard> const&) @ 0x5dfd2c7 in /usr/bin/clickhouse
2020.09.24 13:54:06.588043 [ 17193 ] {} <Fatal> BaseDaemon: 8. ThreadPoolImpl<ThreadFromGlobalPool>::worker(std::__1::__list_iterator<ThreadFromGlobalPool, void*>) @ 0x5d0d06d in /usr/bin/clickhouse
2020.09.24 13:54:06.588105 [ 17193 ] {} <Fatal> BaseDaemon: 9. ThreadFromGlobalPool::ThreadFromGlobalPool<void ThreadPoolImpl<ThreadFromGlobalPool>::scheduleImpl<void>(std::__1::function<void ()>, int, std::__1::optional<unsigned long>)::'lambda1'()>(void&&, void ThreadPoolImpl<ThreadFromGlobalPool>::scheduleImpl<void>(std::__1::function<void ()>, int, std::__1::optional<unsigned long>)::'lambda1'()&&...)::'lambda'()::operator()() const @ 0x5d0d753 in /usr/bin/clickhouse
2020.09.24 13:54:06.588134 [ 17193 ] {} <Fatal> BaseDaemon: 10. ThreadPoolImpl<std::__1::thread>::worker(std::__1::__list_iterator<std::__1::thread, void*>) @ 0x5d0c60d in /usr/bin/clickhouse
2020.09.24 13:54:06.588149 [ 17193 ] {} <Fatal> BaseDaemon: 11. ? @ 0x5d0acdf in /usr/bin/clickhouse
2020.09.24 13:54:06.588176 [ 17193 ] {} <Fatal> BaseDaemon: 12. start_thread @ 0x7e65 in /usr/lib64/libpthread-2.17.so
2020.09.24 13:54:06.588195 [ 17193 ] {} <Fatal> BaseDaemon: 13. __clone @ 0xfe88d in /usr/lib64/libc-2.17.so


2020.09.24 13:54:06.587723 [ 17194 ] {} <Fatal> BaseDaemon: ########################################
2020.09.24 13:54:06.587773 [ 17194 ] {} <Fatal> BaseDaemon: (version 20.6.6.7, no build id) (from thread 17192) (no query) Received signal Segmentation fault (11)
2020.09.24 13:54:06.587786 [ 17194 ] {} <Fatal> BaseDaemon: Address: NULL pointer. Access: read. Unknown si_code.
2020.09.24 13:54:06.587808 [ 17194 ] {} <Fatal> BaseDaemon: Stack trace: 0xd2380e1 0x5d048ff 0xa2a53ef 0x5dfc737 0x5dfd2c7 0x5d0d06d 0x5d0d753 0x5d0c60d 0x5d0acdf 0x7f5b07ba5e65 0x7f5b074c288d
2020.09.24 13:54:06.587877 [ 17194 ] {} <Fatal> BaseDaemon: 3. memcpy @ 0xd2380e1 in /usr/bin/clickhouse
2020.09.24 13:54:06.587930 [ 17194 ] {} <Fatal> BaseDaemon: 4. void DB::writeAnyEscapedString<(char)39, false>(char const*, char const*, DB::WriteBuffer&) @ 0x5d048ff in /usr/bin/clickhouse
2020.09.24 13:54:06.587962 [ 17194 ] {} <Fatal> BaseDaemon: 5. DB::DataTypeString::serializeTextQuoted(DB::IColumn const&, unsigned long, DB::WriteBuffer&, DB::FormatSettings const&) const @ 0xa2a53ef in /usr/bin/clickhouse
2020.09.24 13:54:06.587984 [ 17194 ] {} <Fatal> BaseDaemon: 6. DB::ClusterCopier::getShardPartitions(DB::ConnectionTimeouts const&, DB::TaskShard&) @ 0x5dfc737 in /usr/bin/clickhouse
2020.09.24 13:54:06.587995 [ 17194 ] {} <Fatal> BaseDaemon: 7. DB::ClusterCopier::discoverShardPartitions(DB::ConnectionTimeouts const&, std::__1::shared_ptr<DB::TaskShard> const&) @ 0x5dfd2c7 in /usr/bin/clickhouse
2020.09.24 13:54:06.588008 [ 17194 ] {} <Fatal> BaseDaemon: 8. ThreadPoolImpl<ThreadFromGlobalPool>::worker(std::__1::__list_iterator<ThreadFromGlobalPool, void*>) @ 0x5d0d06d in /usr/bin/clickhouse
2020.09.24 13:54:06.588036 [ 17194 ] {} <Fatal> BaseDaemon: 9. ThreadFromGlobalPool::ThreadFromGlobalPool<void ThreadPoolImpl<ThreadFromGlobalPool>::scheduleImpl<void>(std::__1::function<void ()>, int, std::__1::optional<unsigned long>)::'lambda1'()>(void&&, void ThreadPoolImpl<ThreadFromGlobalPool>::scheduleImpl<void>(std::__1::function<void ()>, int, std::__1::optional<unsigned long>)::'lambda1'()&&...)::'lambda'()::operator()() const @ 0x5d0d753 in /usr/bin/clickhouse
2020.09.24 13:54:06.588064 [ 17194 ] {} <Fatal> BaseDaemon: 10. ThreadPoolImpl<std::__1::thread>::worker(std::__1::__list_iterator<std::__1::thread, void*>) @ 0x5d0c60d in /usr/bin/clickhouse
2020.09.24 13:54:06.588080 [ 17194 ] {} <Fatal> BaseDaemon: 11. ? @ 0x5d0acdf in /usr/bin/clickhouse
2020.09.24 13:54:06.588118 [ 17194 ] {} <Fatal> BaseDaemon: 12. start_thread @ 0x7e65 in /usr/lib64/libpthread-2.17.so
2020.09.24 13:54:06.588142 [ 17194 ] {} <Fatal> BaseDaemon: 13. __clone @ 0xfe88d in /usr/lib64/libc-2.17.so
```

The destination cluster consists of single shard / single replica. 
{
    auto zookeeper = context.getZooKeeper();
    task_description_watch_callback = [this] (const Coordination::WatchResponse & response)
    {
        if (response.error != Coordination::Error::ZOK)
            return;
        UInt64 version = ++task_description_version;
        LOG_DEBUG(log, "Task description should be updated, local version {}", version);
    };
    task_description_path = task_zookeeper_path + "/description";
    task_cluster = std::make_unique<TaskCluster>(task_zookeeper_path, working_database_name);
    reloadTaskDescription();
    task_cluster_initial_config = task_cluster_current_config;
    task_cluster->loadTasks(*task_cluster_initial_config);
    context.setClustersConfig(task_cluster_initial_config, task_cluster->clusters_prefix);
    
    task_cluster->random_engine.seed(task_cluster->random_device());
    for (auto & task_table : task_cluster->table_tasks)
    {
        task_table.cluster_pull = context.getCluster(task_table.cluster_pull_name);
        task_table.cluster_push = context.getCluster(task_table.cluster_push_name);
        task_table.initShards(task_cluster->random_engine);
    }
    LOG_DEBUG(log, "Will process {} table tasks", task_cluster->table_tasks.size());
    
    zookeeper->createAncestors(getWorkersPathVersion() + "/");
    zookeeper->createAncestors(getWorkersPath() + "/");
}
template <typename T>
decltype(auto) ClusterCopier::retry(T && func, UInt64 max_tries)