Can not sink with empty Array

[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. I have a MergeTree table for sink data like this

CREATE OR REPLACE TABLE db.`topic_name`
(
    `before.id` Nullable(UInt32),
    `before.name` Nullable(String),
    `before.slug_name` Nullable(String),
    `before.aftership_id`              Array(String),
    `before.carrier_17_id`             Array(String),
    `before.carrier_cs_phone` Nullable(String),
    `before.carrier_id` Nullable(String),
    `before.carrier_language` Nullable(String),
    `before.carrier_logo` Nullable(String),
    `before.carrier_support_languages` Array(String),
    `before.carrier_url` Nullable(String),
    `before.carrier_url_tracking` Nullable(String),
    `before.comments` Nullable(String),
    `before.carrier_countries_iso`     Array(String),
    `before.pattern_regex`             Array(String),
    `before.scraper_id` Nullable(UInt32),
    `before.base_weight` Nullable(DOUBLE),
    `before.cms_url` Nullable(String),
    `before.allowed_concurrent_request_count` Nullable(UInt32),
    `before.postal_scrape` Nullable(Bool),
    `before.carrier_countries`         Array(String),
    `before.updated_date` Nullable(DateTime64(6)),
    `before.is_hidden` Nullable(Bool),
    `before.merchant_parameters` Nullable(String),
    `before.last_calc_rating` Nullable(DateTime64(6)),
    `before.rating` Nullable(String),
    `after.id` Nullable(UInt32),
    `after.name` Nullable(String),
    `after.slug_name` Nullable(String),
    `after.aftership_id`               Array(String),
    `after.carrier_17_id`              Array(String),
    `after.carrier_cs_phone` Nullable(String),
    `after.carrier_id` Nullable(String),
    `after.carrier_language` Nullable(String),
    `after.carrier_logo` Nullable(String),
    `after.carrier_support_languages`  Array(String) CODEC (ZSTD),
    `after.carrier_url` Nullable(String),
    `after.carrier_url_tracking` Nullable(String),
    `after.comments` Nullable(String),
    `after.carrier_countries_iso`      Array(String),
    `after.pattern_regex`              Array(String),
    `after.scraper_id` Nullable(UInt32),
    `after.base_weight` Nullable(DOUBLE),
    `after.cms_url` Nullable(String),
    `after.allowed_concurrent_request_count` Nullable(UInt32),
    `after.postal_scrape` Nullable(Bool),
    `after.carrier_countries`          Array(String),
    `after.updated_date` Nullable(DateTime64(6)),
    `after.is_hidden` Nullable(Bool),
    `after.merchant_parameters` Nullable(String),
    `after.last_calc_rating` Nullable(DateTime64(6)),
    `after.rating` Nullable(String),
    ts_ms                              Int64 CODEC (LZ4),
    source_table                       Nullable(String),
    op                                 LowCardinality(String) CODEC (ZSTD)
)
    ENGINE MergeTree
        ORDER BY tuple();

My kafka connect config:

name: config-name
  credentialsName: ch-credential
  tasksMax: 1
  config:
    topics: topic_name
    ssl: false
    security.protocol: SASL_PLAINTEXT
    behavior.on.null.values: ignore
    behavior.on.error: ignore
    database: db_name
    hostname: db_host
    port: 8123
    transforms: flatten
    transforms.flatten.type: org.apache.kafka.connect.transforms.Flatten$Value
    transforms.flatten.delimiter: .
    value.converter: io.confluent.connect.avro.AvroConverter
    key.converter: io.confluent.connect.avro.AvroConverter
    value.converter.schema.registry.url: schema-registry:8081
    key.converter.schema.registry.url: schema-registry:8081
    value.converter.schemas.enable: true
    exactlyOnce: true
    errors.log.enable: true
    errors.log.include.messages: true
    schemas.enable: false
    RowBinaryWithDefaults: true

It seemly works with all of types, but when I'm testing with Array(String),kafka connect can not sink empty array into ClickHouse table and throw error like this.

Tasks:
  0:
    State:      FAILED
    Worker ID:  connect:8083
    Trace:      org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:611)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:333)
	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:539)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:833)
      Caused by: java.lang.RuntimeException: Number of records: 500
	at com.clickhouse.kafka.connect.util.Utils.handleException(Utils.java:126)
	at com.clickhouse.kafka.connect.sink.ClickHouseSinkTask.put(ClickHouseSinkTask.java:68)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:581)
	... 10 more
      Caused by: java.lang.RuntimeException: Topic: [topic_name], Partition: [0], MinOffset: [377905], MaxOffset: [378033], (QueryId: [b1cbfdfe-66f4-4404-bdda-736e6da42602])
	at com.clickhouse.kafka.connect.sink.processing.Processing.doInsert(Processing.java:68)
	at com.clickhouse.kafka.connect.sink.processing.Processing.doLogic(Processing.java:165)
	at com.clickhouse.kafka.connect.sink.ProxySinkTask.put(ProxySinkTask.java:99)
	at com.clickhouse.kafka.connect.sink.ClickHouseSinkTask.put(ClickHouseSinkTask.java:64)
	... 11 more
      Caused by: java.util.concurrent.ExecutionException: com.clickhouse.client.ClickHouseException: Code: 190. DB::Exception: Elements 'before.aftership_id' and 'before.carrier_17_id' of Nested data structure 'before' (Array columns) have different array sizes. (SIZES_OF_ARRAYS_DONT_MATCH) (version 24.5.3.5 (official build))

	at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:396)
	at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2073)
	at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.doInsertRawBinary(ClickHouseWriter.java:706)
	at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.doInsert(ClickHouseWriter.java:185)
	at com.clickhouse.kafka.connect.sink.processing.Processing.doInsert(Processing.java:66)
	... 14 more
      Caused by: com.clickhouse.client.ClickHouseException: Code: 190. DB::Exception: Elements 'before.aftership_id' and 'before.carrier_17_id' of Nested data structure 'before' (Array columns) have different array sizes. (SIZES_OF_ARRAYS_DONT_MATCH) (version 24.5.3.5 (official build))

	at com.clickhouse.client.ClickHouseException.of(ClickHouseException.java:151)
	at com.clickhouse.client.AbstractClient.lambda$execute$0(AbstractClient.java:275)
	at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1768)
	... 3 more
      Caused by: java.io.IOException: Code: 190. DB::Exception: Elements 'before.aftership_id' and 'before.carrier_17_id' of Nested data structure 'before' (Array columns) have different array sizes. (SIZES_OF_ARRAYS_DONT_MATCH) (version 24.5.3.5 (official build))

	at com.clickhouse.client.http.ApacheHttpConnectionImpl.checkResponse(ApacheHttpConnectionImpl.java:222)
	at com.clickhouse.client.http.ApacheHttpConnectionImpl.post(ApacheHttpConnectionImpl.java:261)
	at com.clickhouse.client.http.ClickHouseHttpClient.send(ClickHouseHttpClient.java:194)
	at com.clickhouse.client.AbstractClient.sendAsync(AbstractClient.java:161)
	at com.clickhouse.client.AbstractClient.lambda$execute$0(AbstractClient.java:273)
	... 4 more

Topics:
  topic_name

Can you help me to resolve this issue?

[Paultagoras]

Hi @phongbui81062 just curious, could you try setting the column type to Nullable?

It's an interesting problem though - if you had a sample of data, that would help reproduce it as well.