I have Kafka engine table, MV and destination MT table.
Kafka table create query:
```
CREATE TABLE abc.Kafka_TestSource (...) 
ENGINE = Kafka 
SETTINGS kafka_broker_list = '...'
, format_avro_schema_registry_url = '...'
, kafka_topic_list = '...'
, kafka_group_name = 'Test_KafkaToCH'
, kafka_format = 'AvroConfluent'
, kafka_num_consumers = 1
, kafka_thread_per_consumer = 0
```

Due to Grafana consumer group stats dashboard it reads the topic with average speed ~11k records per second.
I recreate Kafka table with `kafka_num_consumers = 9` (kafka topic has 9 partitions) and it still reads with the same speed ~11k rps, **system.metrics** also shows that only 1 pool is being used:
```
SELECT
    m.metric,
    m.value
FROM system.metrics AS m
WHERE m.metric LIKE 'Background%'

Query id: 831a9285-e91b-45a4-8a1b-6b536d6860fe

┌─metric──────────────────────────────────┬─value─┐
│ BackgroundMergesAndMutationsPoolTask    │     4 │
│ BackgroundFetchesPoolTask               │     0 │
│ BackgroundCommonPoolTask                │     0 │
│ BackgroundMovePoolTask                  │     0 │
│ BackgroundSchedulePoolTask              │     0 │
│ BackgroundBufferFlushSchedulePoolTask   │     0 │
│ BackgroundDistributedSchedulePoolTask   │     0 │
│ BackgroundMessageBrokerSchedulePoolTask │     1 │
└─────────────────────────────────────────┴───────┘
```

If I set `kafka_thread_per_consumer = 1` then **BackgroundMessageBrokerSchedulePoolTask** in **system.metrics** raises up to 9 and read speed in Grafana raises up to 100 records per second:
![pic](https://github.com/sokolov-alexey/pics/blob/main/ch_kafka.PNG?raw=true)
1. `kafka_num_consumers = 1, kafka_thread_per_consumer = 0`
2. `kafka_num_consumers = 9, kafka_thread_per_consumer = 0`
3. `kafka_num_consumers = 9, kafka_thread_per_consumer = 1`

Why `kafka_num_consumers` doesn't work without `kafka_thread_per_consumer`?

ClickHouse 21.11.4.14.
}
        #undef CHECK_KAFKA_STORAGE_ARGUMENT
        auto num_consumers = kafka_settings->kafka_num_consumers.value;
        auto max_consumers = std::max<uint32_t>(getNumberOfPhysicalCPUCores(), 16);
        if (num_consumers > max_consumers)
        {
            throw Exception(ErrorCodes::BAD_ARGUMENTS, "The number of consumers can not be bigger than {}. "
                            "A single consumer can read any number of partitions. Extra consumers are relatively expensive, "
                            "and using a lot of them can lead to high memory and CPU usage. To achieve better performance "
                            "of getting data from Kafka, consider using a setting kafka_thread_per_consumer=1, "
                            "and ensure you have enough threads in MessageBrokerSchedulePool (background_message_broker_schedule_pool_size). "
                            "See also https:clickhouse.com/docs/integrations/kafka/kafka-table-engine#tuning-performance", max_consumers);
        }
        else if (num_consumers < 1)
        {
            throw Exception("Number of consumers can not be lower than 1", ErrorCodes::BAD_ARGUMENTS);
        }
        if (kafka_settings->kafka_max_block_size.changed && kafka_settings->kafka_max_block_size.value < 1)
        {
            throw Exception("kafka_max_block_size can not be lower than 1", ErrorCodes::BAD_ARGUMENTS);
        }
        if (kafka_settings->kafka_poll_max_batch_size.changed && kafka_settings->kafka_poll_max_batch_size.value < 1)
        {
            throw Exception("kafka_poll_max_batch_size can not be lower than 1", ErrorCodes::BAD_ARGUMENTS);
        }
        return StorageKafka::create(args.table_id, args.getContext(), args.columns, std::move(kafka_settings), collection_name);
    };
    factory.registerStorage("Kafka", creator_fn, StorageFactory::StorageFeatures{ .supports_settings = true, });
}
NamesAndTypesList StorageKafka::getVirtuals() const