How to deal this issue about kafka sink clickhouse?

[yuanli53]

When I setup posgreSQL to kafka to Clickhouse with this doc:https://github.com/Altinity/clickhouse-sink-connector/blob/develop/doc/quickstart_kafka.md, I got a issue in sink container logs below, Can you help me how to solve this Issue?Thanks.

2024-05-28 05:16:52,223 INFO || preCommit(task-7) 2 [com.altinity.clickhouse.sink.connector.ClickHouseSinkTask]
2024-05-28 05:16:52,233 INFO || close(task-7):2 [com.altinity.clickhouse.sink.connector.ClickHouseSinkTask]
2024-05-28 05:16:52,241 ERROR || WorkerSinkTask{id=sink-connector-7} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted [org.apache.kafka.connect.runtime.WorkerTask]
org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:223)
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:149)
at org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:516)
at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:493)
at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:332)
at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:234)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:203)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:188)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:243)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: org.apache.avro.SchemaParseException: No type: {"content":"{"type":"record","name":"Envelope","namespace":"pg.public.Album","fields":[{"name":"before","type":["null",{"type":"record","name":"Value","fields":[{"name":"AlbumId","type":"int"},{"name":"Title","type":"string"},{"name":"ArtistId","type":"int"}],"connect.name":"pg.public.Album.Value"}],"default":null},{"name":"after","type":["null","Value"],"default":null},{"name":"source","type":"io.debezium.connector.postgresql.Source"},{"name":"op","type":"string"},{"name":"ts_ms","type":["null","long"],"default":null},{"name":"transaction","type":["null",{"type":"record","name":"block","namespace":"event","fields":[{"name":"id","type":"string"},{"name":"total_order","type":"long"},{"name":"data_collection_order","type":"long"}],"connect.version":1,"connect.name":"event.block"}],"default":null}],"connect.version":1,"connect.name":"pg.public.Album.Envelope"}","references":[{"artifactId":"io.debezium.connector.postgresql.Source","version":"1","name":"io.debezium.connector.postgresql.Source"}]}
at org.apache.avro.Schema.getRequiredText(Schema.java:1798)
at org.apache.avro.Schema.parse(Schema.java:1647)
at org.apache.avro.Schema$Parser.parse(Schema.java:1433)
at org.apache.avro.Schema$Parser.parse(Schema.java:1421)
at io.apicurio.registry.serde.avro.AvroSchemaUtils.parse(AvroSchemaUtils.java:55)
at io.apicurio.registry.serde.avro.AvroSchemaParser.parseSchema(AvroSchemaParser.java:43)
at io.apicurio.registry.serde.avro.AvroSchemaParser.parseSchema(AvroSchemaParser.java:28)
at io.apicurio.registry.serde.AbstractSchemaResolver.lambda$resolveSchemaByGlobalId$1(AbstractSchemaResolver.java:179)
at io.apicurio.registry.serde.ERCache.lambda$getValue$0(ERCache.java:132)
at io.apicurio.registry.serde.ERCache.retry(ERCache.java:171)
at io.apicurio.registry.serde.ERCache.getValue(ERCache.java:131)
at io.apicurio.registry.serde.ERCache.getByGlobalId(ERCache.java:111)
at io.apicurio.registry.serde.AbstractSchemaResolver.resolveSchemaByGlobalId(AbstractSchemaResolver.java:172)
at io.apicurio.registry.serde.DefaultSchemaResolver.resolveSchemaByArtifactReference(DefaultSchemaResolver.java:109)
at io.apicurio.registry.serde.AbstractKafkaDeserializer.resolve(AbstractKafkaDeserializer.java:152)
at io.apicurio.registry.serde.AbstractKafkaDeserializer.deserialize(AbstractKafkaDeserializer.java:101)
at io.apicurio.registry.utils.converter.SerdeBasedConverter.toConnectData(SerdeBasedConverter.java:124)
at org.apache.kafka.connect.storage.Converter.toConnectData(Converter.java:88)
at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$4(WorkerSinkTask.java:516)
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:173)
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:207)
... 13 more
2024-05-28 05:16:52,270 INFO || stop(task-7) [com.altinity.clickhouse.sink.connector.ClickHouseSinkTask]