Revert "Send one message to Kafka per row (#284)" (#321)

This reverts commit 7f5f8c2.

Author: qijun-niu-timeplus

src/Storages/ExternalStream/Kafka/KafkaSink.cpp

@@ -223,9 +223,7 @@ KafkaSink::KafkaSink(const Kafka * kafka, const Block & header, ContextPtr conte
     if (data_format.empty())
         data_format = "JSONEachRow";
 
-    /// The callback allows `IRowOutputFormat` based formats produce one Kafka message per row.
-    writer = FormatFactory::instance().getOutputFormat(
-        data_format, *wb, header, context, [this](auto & /*column*/, auto /*row*/) { wb->next(); });
+    writer = FormatFactory::instance().getOutputFormat(data_format, *wb, header, context);
     writer->setAutoFlush();
 
     partitioner = std::make_unique<KafkaStream::ChunkPartitioner>(context, header, kafka->partitioning_expr_ast());

src/Storages/ExternalStream/Kafka/WriteBufferFromKafka.cpp

@@ -20,7 +20,6 @@ void WriteBufferFromKafka::nextImpl()
     if (!offset())
         return;
 
-    /// rd_kafka_produce only enqueue messages, it does not send messages out right away.
     auto err = rd_kafka_produce(
         topic,
         /// we want to trigger the partitioner function, check KafkaSink.cpp