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

break;
                    case PartitionCommand::MoveDestinationType::VOLUME:
                        movePartitionToVolume(command.partition, command.move_destination_name, command.part, query_context);
                        break;
                    case PartitionCommand::MoveDestinationType::TABLE:
                    {
                        checkPartitionCanBeDropped(command.partition);
                        String dest_database = query_context->resolveDatabase(command.to_database);
                        auto dest_storage = DatabaseCatalog::instance().getTable({dest_database, command.to_table}, query_context);
                        movePartitionToTable(dest_storage, command.partition, query_context);
                    }
                    break;
                    case PartitionCommand::MoveDestinationType::SHARD:
                    {
                        if (!getSettings()->part_moves_between_shards_enable)
                            throw Exception(ErrorCodes::SUPPORT_IS_DISABLED,
                                            "Moving parts between shards is experimental and work in progress"
                                            ", see part_moves_between_shards_enable setting");
                        movePartitionToShard(command.partition, command.part, command.move_destination_name, query_context);
                    }
                    break;
                }
            }
            break;
            case PartitionCommand::REPLACE_PARTITION:
            {
                checkPartitionCanBeDropped(command.partition);
                String from_database = query_context->resolveDatabase(command.from_database);
                auto from_storage = DatabaseCatalog::instance().getTable({from_database, command.from_table}, query_context);
                replacePartitionFrom(from_storage, command.partition, command.replace, query_context);