Symptoms: all threads in BackgroundScheduledPool busy, lot of tasks in replication queue, none of the tasks is executing. 

![Screenshot from 2021-04-16 16-21-44_](https://user-images.githubusercontent.com/1549571/115193796-e767d600-a0ec-11eb-887c-5c9c2412e50c.png)

Related to #16895

Possible workarounds (poor options): increasing the background_schedule_pool_size to huge values, or restart server every time when that happens etc. - 

{
    auto zookeeper = getZooKeeper();
    
    zookeeper->createIfNotExists(zookeeper_path + "/quorum", String());
    zookeeper->createIfNotExists(zookeeper_path + "/quorum/parallel", String());
    zookeeper->createIfNotExists(zookeeper_path + "/quorum/last_part", String());
    zookeeper->createIfNotExists(zookeeper_path + "/quorum/failed_parts", String());
    
    zookeeper->createIfNotExists(replica_path + "/min_unprocessed_insert_time", String());
    zookeeper->createIfNotExists(replica_path + "/max_processed_insert_time", String());
    
    zookeeper->createIfNotExists(zookeeper_path + "/mutations", String());
    zookeeper->createIfNotExists(replica_path + "/mutation_pointer", String());
    
    if (storage_settings.get()->allow_s3_zero_copy_replication)
    {
        zookeeper->createIfNotExists(zookeeper_path + "/zero_copy_s3", String());
        zookeeper->createIfNotExists(zookeeper_path + "/zero_copy_s3/shared", String());
    }
    
    zookeeper->createIfNotExists(zookeeper_path + "/part_moves_shard", String());
    zookeeper->createIfNotExists(zookeeper_path + "/pinned_part_uuids", getPinnedPartUUIDs()->toString());
    
    zookeeper->createIfNotExists(zookeeper_path + "/alter_partition_version", String());
}
bool StorageReplicatedMergeTree::createTableIfNotExists(const StorageMetadataPtr & metadata_snapshot)