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 metadata_snapshot = getInMemoryMetadataPtr();
    size_t fields_count = metadata_snapshot->getPartitionKey().sample_block.columns();
    if (partition_ast.fields_count != fields_count)
        throw Exception(
            "Wrong number of fields in the partition expression: " + toString(partition_ast.fields_count) +
            ", must be: " + toString(fields_count),
            ErrorCodes::INVALID_PARTITION_VALUE);
    const FormatSettings format_settings;
    Row partition_row(fields_count);
    if (fields_count)
    {
        ReadBufferFromMemory left_paren_buf("(", 1);
        ReadBufferFromMemory fields_buf(partition_ast.fields_str.data(), partition_ast.fields_str.size());
        ReadBufferFromMemory right_paren_buf(")", 1);
        ConcatReadBuffer buf({&left_paren_buf, &fields_buf, &right_paren_buf});
        auto input_format = FormatFactory::instance().getInput(
            "Values",
            buf,
            metadata_snapshot->getPartitionKey().sample_block,
            local_context,
            local_context->getSettingsRef().max_block_size);
        auto input_stream = std::make_shared<InputStreamFromInputFormat>(input_format);
        auto block = input_stream->read();
        if (!block || !block.rows())
            throw Exception(
                "Could not parse partition value: `" + partition_ast.fields_str + "`",
                ErrorCodes::INVALID_PARTITION_VALUE);
        for (size_t i = 0; i < fields_count; ++i)
            block.getByPosition(i).column->get(0, partition_row[i]);
    }