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.
const auto topic_config_key = config_prefix + "_" + topic;
        if (config.has(topic_config_key))
            loadFromConfig(conf, config, topic_config_key);
    }
    
    conf.set_log_callback([this](cppkafka::KafkaHandleBase &, int level, const std::string & facility, const std::string & message)
    {
        auto [poco_level, client_logs_level] = parseSyslogLevel(level);
        LOG_IMPL(log, client_logs_level, poco_level, "[rdk:{}] {}", facility, message);
    });
    
    
    
    
    {
        
        void * self = static_cast<void *>(this);
        int status;
        status = rd_kafka_conf_interceptor_add_on_new(conf.get_handle(),
            "init", StorageKafkaInterceptors::rdKafkaOnNew, self);
        if (status != RD_KAFKA_RESP_ERR_NO_ERROR)
            LOG_ERROR(log, "Cannot set new interceptor due to {} error", status);
        
        status = rd_kafka_conf_interceptor_add_on_conf_dup(conf.get_handle(),
            "init", StorageKafkaInterceptors::rdKafkaOnConfDup, self);
        if (status != RD_KAFKA_RESP_ERR_NO_ERROR)
            LOG_ERROR(log, "Cannot set dup conf interceptor due to {} error", status);
    }
}
bool StorageKafka::checkDependencies(const StorageID & table_id)