Error sink message from Kafka to table Clickhouse

[phongbui81062]

Hi everyone,

I have this case and I don't know what is the problem.

I create kafka connect sink, which reads data from CDC (Postgres) with Avro format. It worked well but when I add new column in the postgres table. I created new column in MergeTree table, ReplacingMergeTree table and update Materialize view, then I insert new row in postgres table, It occurred this error

Caused by: com.clickhouse.client.ClickHouseException: Code: 33. DB::Exception: Cannot read all data. Bytes read: 180. Bytes expected: 1650954.: (at row 1)

However, when I delete and recreate connection, it worked well. Can you guys help me to explain why it got error like that.

Log from ClickHouse

Code: 33. DB::Exception: Cannot read all data. Bytes read: 174. Bytes expected: 211322150.: (at row 1)
: While executing BinaryRowInputFormat. (CANNOT_READ_ALL_DATA) (version 23.12.2.59 (official build))

0. DB::Exception::Exception(DB::Exception::MessageMasked&&, int, bool) @ 0x000000000bae58c8 in /usr/bin/clickhouse
1. DB::Exception::Exception<unsigned long&, String>(int, FormatStringHelperImpl<std::type_identity<unsigned long&>::type, std::type_identity<String>::type>, unsigned long&, String&&) @ 0x000000000751f778 in /usr/bin/clickhouse
2. DB::ReadBuffer::readStrict(char*, unsigned long) @ 0x000000000751f668 in /usr/bin/clickhouse
3. DB::SerializationString::deserializeBinary(DB::IColumn&, DB::ReadBuffer&, DB::FormatSettings const&) const @ 0x000000000f4752e4 in /usr/bin/clickhouse
4. DB::SerializationNullable::deserializeBinary(DB::IColumn&, DB::ReadBuffer&, DB::FormatSettings const&) const @ 0x000000000f464188 in /usr/bin/clickhouse
5. DB::BinaryFormatReader<false>::readField(DB::IColumn&, std::shared_ptr<DB::IDataType const> const&, std::shared_ptr<DB::ISerialization const> const&, bool, String const&) @ 0x0000000010ff2f28 in /usr/bin/clickhouse
6. DB::RowInputFormatWithNamesAndTypes::readRow(std::vector<COW<DB::IColumn>::mutable_ptr<DB::IColumn>, std::allocator<COW<DB::IColumn>::mutable_ptr<DB::IColumn>>>&, DB::RowReadExtension&) @ 0x0000000010ff7190 in /usr/bin/clickhouse
7. DB::IRowInputFormat::generate() @ 0x0000000010fc5b10 in /usr/bin/clickhouse
8. DB::ISource::tryGenerate() @ 0x0000000010f4952c in /usr/bin/clickhouse
9. DB::ISource::work() @ 0x0000000010f48f54 in /usr/bin/clickhouse
10. DB::ExecutionThreadContext::executeTask() @ 0x0000000010f5d38c in /usr/bin/clickhouse
11. DB::PipelineExecutor::executeStepImpl(unsigned long, std::atomic<bool>*) @ 0x0000000010f55574 in /usr/bin/clickhouse
12. DB::PipelineExecutor::execute(unsigned long, bool) @ 0x0000000010f54b80 in /usr/bin/clickhouse
13. DB::CompletedPipelineExecutor::execute() @ 0x0000000010f53518 in /usr/bin/clickhouse
14. DB::executeQuery(DB::ReadBuffer&, DB::WriteBuffer&, bool, std::shared_ptr<DB::Context>, std::function<void (DB::QueryResultDetails const&)>, DB::QueryFlags, std::optional<DB::FormatSettings> const&, std::function<void (DB::IOutputFormat&)>) @ 0x00000000101fd44c in /usr/bin/clickhouse
15. DB::HTTPHandler::processQuery(DB::HTTPServerRequest&, DB::HTMLForm&, DB::HTTPServerResponse&, DB::HTTPHandler::Output&, std::optional<DB::CurrentThread::QueryScope>&) @ 0x0000000010eab184 in /usr/bin/clickhouse
16. DB::HTTPHandler::handleRequest(DB::HTTPServerRequest&, DB::HTTPServerResponse&) @ 0x0000000010eaf004 in /usr/bin/clickhouse
17. DB::HTTPServerConnection::run() @ 0x0000000010f1658c in /usr/bin/clickhouse
18. Poco::Net::TCPServerConnection::start() @ 0x00000000133a9a84 in /usr/bin/clickhouse
19. Poco::Net::TCPServerDispatcher::run() @ 0x00000000133aac20 in /usr/bin/clickhouse
20. Poco::PooledThread::run() @ 0x00000000134cc7fc in /usr/bin/clickhouse
21. Poco::ThreadImpl::runnableEntry(void*) @ 0x00000000134cad44 in /usr/bin/clickhouse
22. start_thread @ 0x0000000000007624 in /usr/lib/aarch64-linux-gnu/libpthread-2.31.so
23. ? @ 0x00000000000d162c in /usr/lib/aarch64-linux-gnu/libc-2.31.so

[Paultagoras]

Did you restart the connector at all after creating the new column? We describe the table in the beginning so new fields aren't added until a reboot.

[phongbui81062]

Thank for your answer

[phongbui81062]

Hi @Paultagoras
Is there any option that help me to choose specific schema registry version?

[Paultagoras]

Uh, unsure what you mean - Kafka Connect has a configuration for schema registry, of course, but usually the registry itself is provided as part of whichever platform (Confluent, MSK, etc...)