(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)


auto consumer = std::make_shared<cppkafka::Consumer>(conf);
    consumer->set_destroy_flags(RD_KAFKA_DESTROY_F_NO_CONSUMER_CLOSE);
    
    if (thread_per_consumer)
    {
        auto& stream_cancelled = tasks[consumer_number]->stream_cancelled;
        return std::make_shared<ReadBufferFromKafkaConsumer>(consumer, log, getPollMaxBatchSize(), getPollTimeoutMillisecond(), intermediate_commit, stream_cancelled, topics);
    }
    return std::make_shared<ReadBufferFromKafkaConsumer>(consumer, log, getPollMaxBatchSize(), getPollTimeoutMillisecond(), intermediate_commit, tasks.back()->stream_cancelled, topics);
}
size_t StorageKafka::getMaxBlockSize() const
{
    return kafka_settings->kafka_max_block_size.changed
        ? kafka_settings->kafka_max_block_size.value
        : (getContext()->getSettingsRef().max_insert_block_size.value / num_consumers);
}
size_t StorageKafka::getPollMaxBatchSize() const
{
    size_t batch_size = kafka_settings->kafka_poll_max_batch_size.changed
                        ? kafka_settings->kafka_poll_max_batch_size.value
                        : getContext()->getSettingsRef().max_block_size.value;
    return std::min(batch_size,getMaxBlockSize());
}
size_t StorageKafka::getPollTimeoutMillisecond() const
{
    return kafka_settings->kafka_poll_timeout_ms.changed
        ? kafka_settings->kafka_poll_timeout_ms.totalMilliseconds()
        : getContext()->getSettingsRef().stream_poll_timeout_ms.totalMilliseconds();
}
void StorageKafka::updateConfiguration(cppkafka::Configuration & conf)