Certain characters in a table column comment can break the connector

[ne1r0n]

Describe the bug

It was found out that describeTableV1 breaks if there is a \ character in the comment of a table column. Connector stops working.

Steps to reproduce

1. Create a table

CREATE TABLE comment
(
  c String COMMENT '\\'
)
ENGINE = MergeTree()
ORDER BY tuple()

2. Run connector with V1 client_version

Expected behaviour

The column comment does not affect the operation of the connector.

Error log

ERROR [ch-kt-connector-dev|task-0] Exception when running describeTable DESCRIBE TABLE `test`.`comment` (com.clickhouse.kafka.connect.sink.db.helper.ClickHouseHelperClient:386)
com.fasterxml.jackson.core.JsonParseException: Unexpected character ('c' (code 99)): was expecting comma to separate Object entries
 at [Source: REDACTED (`StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION` disabled); line: 1, column: 86]
	at com.fasterxml.jackson.core.JsonParser._constructReadException(JsonParser.java:2660)
	at com.fasterxml.jackson.core.base.ParserMinimalBase._reportUnexpectedChar(ParserMinimalBase.java:741)
	at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._skipComma(ReaderBasedJsonParser.java:2429)
	at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.nextToken(ReaderBasedJsonParser.java:691)
	at com.fasterxml.jackson.databind.deser.BuilderBasedDeserializer.vanillaDeserialize(BuilderBasedDeserializer.java:287)
	at com.fasterxml.jackson.databind.deser.BuilderBasedDeserializer.deserialize(BuilderBasedDeserializer.java:218)
	at com.fasterxml.jackson.databind.deser.DefaultDeserializationContext.readRootValue(DefaultDeserializationContext.java:342)
	at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4917)
	at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3860)
	at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3828)
	at com.clickhouse.kafka.connect.sink.db.helper.ClickHouseFieldDescriptor.fromJsonRow(ClickHouseFieldDescriptor.java:50)
	at com.clickhouse.kafka.connect.sink.db.helper.ClickHouseHelperClient.describeTableV1(ClickHouseHelperClient.java:366)
	at com.clickhouse.kafka.connect.sink.db.helper.ClickHouseHelperClient.describeTable(ClickHouseHelperClient.java:343)
	at com.clickhouse.kafka.connect.sink.db.helper.ClickHouseHelperClient.extractTablesMapping(ClickHouseHelperClient.java:447)
	at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.updateMapping(ClickHouseWriter.java:156)
	at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.start(ClickHouseWriter.java:136)
	at com.clickhouse.kafka.connect.sink.ProxySinkTask.<init>(ProxySinkTask.java:64)
	at com.clickhouse.kafka.connect.sink.ClickHouseSinkTask.start(ClickHouseSinkTask.java:42)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.initializeAndStart(WorkerSinkTask.java:324)
	at org.apache.kafka.connect.runtime.WorkerTask.doStart(WorkerTask.java:176)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:225)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:281)
	at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:238)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
	at java.base/java.util.concurrent.FutureTask.run(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
	at java.base/java.lang.Thread.run(Unknown Source)

Configuration

Environment

• Kafka-Connect version: 1.2.9

[Paultagoras]

@ne1r0n Will take a look - thanks!