(you don't have to strictly follow this form)

**Use case**
Our Machine is 96 cores, 256 GB, 10T ssd.
We want to use the Kafka Engine table and distribute table only for writing data to other worker nodes .

**Describe the solution you'd like**
One Kafka Engine Node can consumer more data, and reduce the other ch node work load.

other worker node can get more performance. 

![image](https://user-images.githubusercontent.com/77636494/126424731-5883f321-a8d5-4954-a481-0ea3000db933.png)


* - 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
          */
        /* 
        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)
        #undef CHECK_KAFKA_STORAGE_ARGUMENT
        auto num_consumers = kafka_settings->kafka_num_consumers.value;
        if (num_consumers > 16)
        {
            throw Exception("Number of consumers can not be bigger than 16", ErrorCodes::BAD_ARGUMENTS);
        }
        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)
        {