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.
/** Arguments of engine is following:
          * - Kafka broker list
          * - List of topics
          * - Group ID (may be a constaint expression with a string result)
          * - Message format (string)
          * - Row delimiter
          * - Schema (optional, if the format supports it)
          * - Number of consumers
          * - Max block size for background consumption
          * - Skip (at least) unreadable messages number
          * - Do intermediate commits when the batch consumed and handled
          */
        String collection_name;
        if (named_collection)
        {
            collection_name = assert_cast<const ASTIdentifier *>(args.engine_args[0].get())->name();
        }
        else
        {
            /* 
            CHECK_KAFKA_STORAGE_ARGUMENT(1, kafka_broker_list, 0)
            CHECK_KAFKA_STORAGE_ARGUMENT(2, kafka_topic_list, 1)
            CHECK_KAFKA_STORAGE_ARGUMENT(3, kafka_group_name, 2)
            CHECK_KAFKA_STORAGE_ARGUMENT(4, kafka_format, 2)
            CHECK_KAFKA_STORAGE_ARGUMENT(5, kafka_row_delimiter, 2)
            CHECK_KAFKA_STORAGE_ARGUMENT(6, kafka_schema, 2)
            CHECK_KAFKA_STORAGE_ARGUMENT(7, kafka_num_consumers, 0)
            CHECK_KAFKA_STORAGE_ARGUMENT(8, kafka_max_block_size, 0)
            CHECK_KAFKA_STORAGE_ARGUMENT(9, kafka_skip_broken_messages, 0)
            CHECK_KAFKA_STORAGE_ARGUMENT(10, kafka_commit_every_batch, 0)