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 (required_columns.empty() && !keys.empty())
                continue;
            if (analysis_result.optimize_aggregation_in_order)
            {
                for (const auto & key : keys)
                {
                    auto actions_dag = analysis_result.before_aggregation->clone();
                    actions_dag->foldActionsByProjection({key}, projection.sample_block_for_keys);
                    candidate.group_by_elements_actions.emplace_back(std::make_shared<ExpressionActions>(actions_dag, actions_settings));
                }
            }
            /
            candidate.before_aggregation->reorderAggregationKeysForProjection(key_name_pos_map);
            candidate.before_aggregation->addAggregatesViaProjection(aggregates);
            if (rewrite_before_where(candidate, projection, required_columns, projection.sample_block_for_keys, aggregates))
            {
                candidate.required_columns = {required_columns.begin(), required_columns.end()};
                for (const auto & aggregate : aggregates)
                    candidate.required_columns.push_back(aggregate.name);
                candidates.push_back(std::move(candidate));
            }
            /
        }
        if (projection.type == ProjectionDescription::Type::Normal && (analysis_result.hasWhere() || analysis_result.hasPrewhere()))
        {
            /
            const auto & actions
                = analysis_result.before_aggregation ? analysis_result.before_aggregation : analysis_result.before_order_by;
            NameSet required_columns;
            for (const auto & column : actions->getRequiredColumns())