When using protobuf, how to define UUID and UInt64 in the proto file?

[danifullview]

We're trying to switch from json to protobuf and we're encountering the following error.

When creating the ClickHouse table, we're using UUIDs and UInt64.
When defining the protobuf model, we're using String for UUID and uint64 for UInt64.

The Sink Connector throws these errors when processing the messages:

ERROR Table column name [id] type [NONE] is not matching data column type [STRING] (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter)
ERROR Table column name [timestamp] type [NONE] is not matching data column type [INT64] (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter)

Here's the stack trace:

java.lang.RuntimeException: java.lang.NullPointerException
	at com.clickhouse.kafka.connect.util.Utils.handleException(Utils.java:86)
	at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.doInsert(ClickHouseWriter.java:143)
	at com.clickhouse.kafka.connect.sink.processing.Processing.doInsert(Processing.java:45)
	at com.clickhouse.kafka.connect.sink.processing.Processing.doLogic(Processing.java:120)
	at com.clickhouse.kafka.connect.sink.ProxySinkTask.put(ProxySinkTask.java:80)
	at com.clickhouse.kafka.connect.sink.ClickHouseSinkTask.put(ClickHouseSinkTask.java:60)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:587)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:257)
	at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181)
	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: java.lang.NullPointerException
	at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.validateDataSchema(ClickHouseWriter.java:161)
	at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.doInsertRawBinary(ClickHouseWriter.java:353)
	at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.doInsert(ClickHouseWriter.java:135)

[danifullview]

@Paultagoras could you please give me a helping hand?

[Paultagoras]

Working on a resolution now - it's an interesting issue

[Paultagoras]

Yeah I think we have a fix for this - it's essentially because those datatypes weren't being converted. I'll create a PR for the fix - it should be in the next release coming today

[danifullview]

Thank you @Paultagoras.
I tried your code and it works.

There's another issue though.

Having the schema above, if I don't provide the id, the sink connector crashes w/ the following exception:

[2023-06-16 09:22:14,443] WARN ClickHouseException code: 27 (com.clickhouse.kafka.connect.util.Utils)
[2023-06-16 09:22:14,443] ERROR Error code [27] wasn't in the acceptable list. (com.clickhouse.kafka.connect.util.Utils)
[2023-06-16 09:22:14,444] ERROR WorkerSinkTask{id=ClickHouseProtoSinkConnector-1} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted. Error: java.util.concurrent.ExecutionException: com.clickhouse.client.ClickHouseException: Code: 27. DB::ParsingException: Cannot parse input: expected '"' before: 'a1c24b1e-e762-42d0-907d-ed16ffa510bc","timestamp":1686907332779}': (while reading the value of key id): While executing ParallelParsingBlockInputFormat: (at row 1)
. (CANNOT_PARSE_INPUT_ASSERTION_FAILED) (version 23.5.1.34385 (official build))
, server ClickHouseNode [uri=https://jb38c2mzk2.eu-central-1.aws.clickhouse.cloud:8443/fullview, options={sslmode=STRICT}] (org.apache.kafka.connect.runtime.WorkerSinkTask)
java.lang.RuntimeException: java.util.concurrent.ExecutionException: com.clickhouse.client.ClickHouseException: Code: 27. DB::ParsingException: Cannot parse input: expected '"' before: 'a1c24b1e-e762-42d0-907d-ed16ffa510bc","timestamp":1686907332779}': (while reading the value of key id): While executing ParallelParsingBlockInputFormat: (at row 1)
. (CANNOT_PARSE_INPUT_ASSERTION_FAILED) (version 23.5.1.34385 (official build))
, server ClickHouseNode [uri=https://jb38c2mzk2.eu-central-1.aws.clickhouse.cloud:8443/fullview, options={sslmode=STRICT}]
	at com.clickhouse.kafka.connect.util.Utils.handleException(Utils.java:86)
	at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.doInsert(ClickHouseWriter.java:152)
	at com.clickhouse.kafka.connect.sink.processing.Processing.doInsert(Processing.java:45)
	at com.clickhouse.kafka.connect.sink.processing.Processing.doLogic(Processing.java:164)
	at com.clickhouse.kafka.connect.sink.ProxySinkTask.put(ProxySinkTask.java:80)
	at com.clickhouse.kafka.connect.sink.ClickHouseSinkTask.put(ClickHouseSinkTask.java:60)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:587)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:257)
	at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181)
	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: java.util.concurrent.ExecutionException: com.clickhouse.client.ClickHouseException: Code: 27. DB::ParsingException: Cannot parse input: expected '"' before: 'a1c24b1e-e762-42d0-907d-ed16ffa510bc","timestamp":1686907332779}': (while reading the value of key id): While executing ParallelParsingBlockInputFormat: (at row 1)
. (CANNOT_PARSE_INPUT_ASSERTION_FAILED) (version 23.5.1.34385 (official build))
, server ClickHouseNode [uri=https://jb38c2mzk2.eu-central-1.aws.clickhouse.cloud:8443/fullview, options={sslmode=STRICT}]
	at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395)
	at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1999)
	at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.doInsertJson(ClickHouseWriter.java:519)
	at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.doInsert(ClickHouseWriter.java:141)
	... 16 more

Should I open an issue w/ this?

[Paultagoras]

Is this still an issue you're experiencing in 0.0.16?

[danifullview]

I haven't tried it.
I know the UUID fix worked, I merged your PR and generated the binary locally.

I will try the DEFAULT behaviour and let you know.