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


#include <Storages/StorageFactory.h>
#include <Storages/StorageMaterializedView.h>
#include <boost/algorithm/string/replace.hpp>
#include <boost/algorithm/string/split.hpp>
#include <boost/algorithm/string/trim.hpp>
#include <Poco/Util/AbstractConfiguration.h>
#include <Common/Exception.h>
#include <Common/Macros.h>
#include <Common/formatReadable.h>
#include <Common/config_version.h>
#include <Common/setThreadName.h>
#include <Common/typeid_cast.h>
#include <common/logger_useful.h>
#include <Common/quoteString.h>
#include <Interpreters/Context.h>
#include <Processors/Sources/SourceFromInputStream.h>
#include <librdkafka/rdkafka.h>
#include <common/getFQDNOrHostName.h>
namespace DB
{
namespace ErrorCodes
{
    extern const int NOT_IMPLEMENTED;
    extern const int LOGICAL_ERROR;
    extern const int BAD_ARGUMENTS;
    extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
}
struct StorageKafkaInterceptors
{
    static rd_kafka_resp_err_t rdKafkaOnThreadStart(rd_kafka_t *, rd_kafka_thread_type_t thread_type, const char *, void * ctx)