You have to provide the following information whenever possible.

Describe the bug

Any attempt to insert into a table with engine Kafka (e.g. test_kafka_insert from integration test) causes assertion failure.

Does it reproduce on recent release?

I am playing with recent master c255f15

Error message and/or stacktrace

2021.07.19 23:31:48.695409 [ 1592186 ] {2d87e4e5-4795-4c31-ac04-2ec059dbabc5} MemoryTracker: Peak memory usage (for query): 0.00 B.
clickhouse: /home/ilejn/projects/ClickHouse/src/Storages/Kafka/WriteBufferToKafkaProducer.cpp:51: virtual DB::WriteBufferToKafkaProducer::~WriteBufferToKafkaProducer(): Assertion `rows == 0 && chunks.empty()' failed.
2021.07.19 23:31:48.696445 [ 1592136 ] {} BaseDaemon: Received signal 6
2021.07.19 23:31:48.697249 [ 1592463 ] {} BaseDaemon: ########################################
2021.07.19 23:31:48.697505 [ 1592463 ] {} BaseDaemon: (version 21.7.1.1, no build id) (from thread 1592186) (query_id: 2d87e4e5-4795-4c31-ac04-2ec059dbabc5) Received signal Aborted (6)
2021.07.19 23:31:48.697648 [ 1592463 ] {} BaseDaemon:
2021.07.19 23:31:48.697802 [ 1592463 ] {} BaseDaemon: Stack trace: 0x7f8266f12fbb 0x7f8266ef8864 0x7f8266ef8749 0x7f8266f0a3d6 0x146dd808 0x146d89f9 0x146d89bd 0x146d897d 0x146d862e 0x9980c71 0x9980c19 0x146d8b0c 0x1523639b 0x146d4f7a 0x146d4f3d 0x146d4efd 0x146d4cae 0x9980c71 0x9980c19 0x9a9b8ec 0x1478a648 0x1389f75a 0x1389f71d 0x1389f6dd 0x1389f48e 0x9980c71 0x9980c19 0x9a9b8ec 0x147b1588 0x138a03ba 0x138a037d 0x138a033d 0x138a008e 0x9980c71 0x9980c19 0x9a9b8ec 0x1478f85b 0x138a0e7a 0x138a0e3d 0x138a0dfd 0x138a0bce
2021.07.19 23:31:48.698029 [ 1592463 ] {} BaseDaemon: 4. gsignal @ 0x40fbb in /usr/lib/x86_64-linux-gnu/libc-2.33.so
2021.07.19 23:31:48.698113 [ 1592463 ] {} BaseDaemon: 5. abort @ 0x26864 in /usr/lib/x86_64-linux-gnu/libc-2.33.so
2021.07.19 23:31:48.698188 [ 1592463 ] {} BaseDaemon: 6. ? @ 0x26749 in /usr/lib/x86_64-linux-gnu/libc-2.33.so
2021.07.19 23:31:48.698263 [ 1592463 ] {} BaseDaemon: 7. ? @ 0x383d6 in /usr/lib/x86_64-linux-gnu/libc-2.33.so
2021.07.19 23:31:48.824976 [ 1592463 ] {} BaseDaemon: 8. /home/ilejn/projects/ClickHouse/src/Storages/Kafka/WriteBufferToKafkaProducer.cpp:0: DB::WriteBufferToKafkaProducer::~WriteBufferToKafkaProducer() @ 0x146dd808 in /home/ilejn/projects/ClickHouse/build/programs/clickhouse
2021.07.19 23:31:49.466935 [ 1592463 ] {} BaseDaemon: 9. /home/ilejn/projects/ClickHouse/contrib/libcxx/include/memory:892: std::__1::allocatorDB::WriteBufferToKafkaProducer::destroy(DB::WriteBufferToKafkaProducer*) @ 0x146d89f9 in /home/ilejn/projects/ClickHouse/build/programs/clickhouse
2021.07.19 23:31:50.229230 [ 1592463 ] {} BaseDaemon: 10. /home/ilejn/projects/ClickHouse/contrib/libcxx/include/__memory/allocator_traits.h:541: void std::__1::allocator_traits<std::__1::allocatorDB::WriteBufferToKafkaProducer >::__destroyDB::WriteBufferToKafkaProducer(std::__1::integral_constant<bool, true>, std::__1::allocatorDB::WriteBufferToKafkaProducer&, DB::WriteBufferToKafkaProducer*) @ 0x146d89bd in /home/ilejn/projects/ClickHouse/build/programs/clickhouse
2021.07.19 23:31:50.877117 [ 1592463 ] {} BaseDaemon: 11. /home/ilejn/projects/ClickHouse/contrib/libcxx/include/__memory/allocator_traits.h:487: void std::__1::allocator_traits<std::__1::allocatorDB::WriteBufferToKafkaProducer >::destroyDB::WriteBufferToKafkaProducer(std::__1::allocatorDB::WriteBufferToKafkaProducer&, DB::WriteBufferToKafkaProducer*) @ 0x146d897d in /home/ilejn/projects/ClickHouse/build/programs/clickhouse
2021.07.19 23:31:51.536775 [ 1592463 ] {} BaseDaemon: 12. /home/ilejn/projects/ClickHouse/contrib/libcxx/include/memory:2611: std::__1::__shared_ptr_emplace<DB::WriteBufferToKafkaProducer, std::__1::allocatorDB::WriteBufferToKafkaProducer >::__on_zero_shared() @ 0x146d862e in /home/ilejn/projects/ClickHouse/build/programs/clickhouse
2021.07.19 23:31:51.731367 [ 1592463 ] {} BaseDaemon: 13. /home/ilejn/projects/ClickHouse/contrib/libcxx/include/memory:2476: std::__1::__shared_count::__release_shared() @ 0x9980c71 in /home/ilejn/projects/ClickHouse/build/programs/clickhouse
2021.07.19 23:31:51.909613 [ 1592463 ] {} BaseDaemon: 14. /home/ilejn/projects/ClickHouse/contrib/libcxx/include/memory:2517: std::__1::__shared_weak_count::__release_shared() @ 0x9980c19 in /home/ilejn/projects/ClickHouse/build/programs/clickhouse
2021.07.19 23:31:52.532413 [ 1592463 ] {} BaseDaemon: 15. /home/ilejn/projects/ClickHouse/contrib/libcxx/include/memory:3213: std::__1::shared_ptrDB::WriteBufferToKafkaProducer::~shared_ptr() @ 0x146d8b0c in /home/ilejn/projects/ClickHouse/build/programs/clickhouse
2021.07.19 23:31:52.739467 [ 1592463 ] {} BaseDaemon: 16. /home/ilejn/projects/ClickHouse/src/Storages/Kafka/KafkaBlockOutputStream.h:9: DB::KafkaBlockOutputStream::~KafkaBlockOutputStream() @ 0x1523639b in /home/ilejn/projects/ClickHouse/build/programs/clickhouse