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


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)
{
    
    auto dependencies = DatabaseCatalog::instance().getDependencies(table_id);
    if (dependencies.empty())
        return true;
    
    for (const auto & db_tab : dependencies)
    {
        auto table = DatabaseCatalog::instance().tryGetTable(db_tab, getContext());
        if (!table)
            return false;
        
        auto * materialized_view = dynamic_cast<StorageMaterializedView *>(table.get());
        if (materialized_view && !materialized_view->tryGetTargetTable())
            return false;
        
        if (!checkDependencies(db_tab))
            return false;
    }
    return true;
}
void StorageKafka::threadFunc(size_t idx)