Clickhouse Error DB::Exception: Pos in empty own buffer, Stack trace (when copying this message, always include the lines below)

Question

Could anyone help me with this error? I don't understand what it means and how to fix it. Clickhouse version 21.7.2.7

I have several partitions in Kafka. When I create a Kafka Engine table and attach a materialized view to it, I've got this type of error. But interesting behavior is that the Kafka engine continues to ingest data from some partitions and skip another.

<Error> void DB::StorageKafka::threadFunc(size_t): Code: 49, e.displayText() = DB::Exception: Pos in empty own buffer, Stack trace (when copying this message, always include the lines below):
0. DB::Exception::Exception(std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&, int, bool) @ 0x8d35c1a in /usr/bin/clickhouse
1. DB::PeekableReadBuffer::checkStateCorrect() const @ 0x106bc556 in /usr/bin/clickhouse
2. DB::JSONAsStringRowInputFormat::resetParser() @ 0x1078adf3 in /usr/bin/clickhouse
3. DB::KafkaBlockInputStream::readImpl() @ 0x1026fd2d in /usr/bin/clickhouse
4. DB::IBlockInputStream::read() @ 0xf51faa4 in /usr/bin/clickhouse
5. DB::copyData(DB::IBlockInputStream&, DB::IBlockOutputStream&, std::__1::function<void (DB::Block const&)> const&, std::__1::atomic<bool>*) @ 0xf542cf5 in /usr/bin/clickhouse
6. DB::StorageKafka::streamToViews() @ 0x10264279 in /usr/bin/clickhouse
7. DB::StorageKafka::threadFunc(unsigned long) @ 0x10262e58 in /usr/bin/clickhouse
8. DB::BackgroundSchedulePoolTaskInfo::execute() @ 0xf772a40 in /usr/bin/clickhouse
9. DB::BackgroundSchedulePool::threadFunction() @ 0xf774ab7 in /usr/bin/clickhouse
10. ? @ 0xf775834 in /usr/bin/clickhouse
11. ThreadPoolImpl<std::__1::thread>::worker(std::__1::__list_iterator<std::__1::thread, void*>) @ 0x8d76adf in /usr/bin/clickhouse
12. ? @ 0x8d7a3c3 in /usr/bin/clickhouse
13. start_thread @ 0x9609 in /usr/lib/x86_64-linux-gnu/libpthread-2.31.so
14. __clone @ 0x122293 in /usr/lib/x86_64-linux-gnu/libc-2.31.so
 (version 21.7.2.7 (official build))

Kafka engine

CREATE TABLE IF NOT EXISTS db.kafka_table
(
    `message` String
)
ENGINE = Kafka
SETTINGS kafka_broker_list = 'host:29092,host2:29092,host3:29092',
    kafka_topic_list = 'topic',
    kafka_group_name = 'group',
    kafka_format = 'JSONAsString',
    kafka_num_consumers = 3,
    kafka_max_block_size = 10000,
    kafka_poll_max_batch_size = 10000,
    kafka_thread_per_consumer = 1,
    kafka_handle_error_mode = 'stream';

Answer 1

I had a problem with data that had been placed in Kafka. It looked like

"{\r\n\t\"field_1\": \"value_1\", ..., \r\n\t\"field_n\": \"value_n\"\r\n}"

Hence, Clickhouse Kafka Engine couldn't parse it and commit offset, so this was a cause of big consumer lag. Changing the input format has fixed the problem but I have to add supplementary JSON validation in materialized view attached to the Kafka engine table.

correct SQL

CREATE TABLE IF NOT EXISTS db.kafka_table
(
    `message` String
)
ENGINE = Kafka
SETTINGS kafka_broker_list = 'host:29092,host2:29092,host3:29092',
    kafka_topic_list = 'topic',
    kafka_group_name = 'group',
    kafka_format = 'RawBLOB',
    kafka_num_consumers = 3,
    kafka_max_block_size = 10000,
    kafka_poll_max_batch_size = 10000,
    kafka_thread_per_consumer = 1,
    kafka_handle_error_mode = 'stream';