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. - 

if (!source_part_or_covering->info.contains(MergeTreePartInfo::fromPartName(entry.new_part_name, format_version)))
                throw Exception(ErrorCodes::LOGICAL_ERROR, message, source_part_name, source_part_or_covering->name, entry.new_part_name);
            return false;
        }
        parts.push_back(source_part_or_covering);
    }
    
    if (entry.create_time + storage_settings_ptr->prefer_fetch_merged_part_time_threshold.totalSeconds() <= time(nullptr))
    {
        
        
        size_t sum_parts_bytes_on_disk = 0;
        for (const auto & part : parts)
            sum_parts_bytes_on_disk += part->getBytesOnDisk();
        if (sum_parts_bytes_on_disk >= storage_settings_ptr->prefer_fetch_merged_part_size_threshold)
        {
            String replica = findReplicaHavingPart(entry.new_part_name, true);    
            if (!replica.empty())
            {
                LOG_DEBUG(log, "Prefer to fetch {} from replica {}", entry.new_part_name, replica);
                return false;
            }
        }
    }
    
    size_t estimated_space_for_merge = MergeTreeDataMergerMutator::estimateNeededDiskSpace(parts);
    
    IMergeTreeDataPart::TTLInfos ttl_infos;
    size_t max_volume_index = 0;
    for (auto & part_ptr : parts)