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


: getContext()->getSettingsRef().stream_flush_interval_ms;
        limits.timeout_overflow_mode = OverflowMode::BREAK;
        stream->setLimits(limits);
    }
    
    BlockInputStreamPtr in;
    if (streams.size() > 1)
        in = std::make_shared<UnionBlockInputStream>(streams, nullptr, streams.size());
    else
        in = streams[0];
    
    
    std::atomic<bool> stub = {false};
    size_t rows = 0;
    copyData(*in, *block_io.out, [&rows](const Block & block)
    {
        rows += block.rows();
    }, &stub);
    bool some_stream_is_stalled = false;
    for (auto & stream : streams)
    {
        some_stream_is_stalled = some_stream_is_stalled || stream->as<KafkaBlockInputStream>()->isStalled();
        stream->as<KafkaBlockInputStream>()->commit();
    }
    UInt64 milliseconds = watch.elapsedMilliseconds();
    LOG_DEBUG(log, "Pushing {} rows to {} took {} ms.",
        formatReadableQuantity(rows), table_id.getNameForLogs(), milliseconds);
    return some_stream_is_stalled;
}
void registerStorageKafka(StorageFactory & factory)