How to control the amount of polled records to avoid small chunk inserts

[abdelhakimbendjabeur]

Hello,
I have been trying to tweak the configuration in order to avoid having small batch inserts (see logs below)

[2023-06-15 10:13:23,500] INFO [analytics-ticket-clickhouse-sink|task-0]  doLogic Topic [analytics_sparse_ticket_dataset] Partition [14] offset [34272377:34272378] (com.clickhouse.kafka.connect.sink.processing.Processing:88)
[2023-06-15 10:13:23,500] INFO [analytics-ticket-clickhouse-sink|task-0]  Number of records to insert 2 to table name analytics_sparse_ticket_dataset (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter:347)
[2023-06-15 10:13:23,565] INFO [analytics-ticket-clickhouse-sink|task-0]  batchSize 2 data ms 0 send 65 (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter:403)
[2023-06-15 10:13:23,817] INFO [analytics-ticket-clickhouse-sink|task-0]  doLogic Topic [analytics_sparse_ticket_dataset] Partition [1] offset [34253092:34253126] (com.clickhouse.kafka.connect.sink.processing.Processing:88)
[2023-06-15 10:13:23,817] INFO [analytics-ticket-clickhouse-sink|task-0]  Number of records to insert 35 to table name analytics_sparse_ticket_dataset (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter:347)
[2023-06-15 10:13:23,879] INFO [analytics-ticket-clickhouse-sink|task-0]  batchSize 35 data ms 0 send 62 (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter:403)
[2023-06-15 10:13:24,455] INFO [analytics-ticket-clickhouse-sink|task-0]  doLogic Topic [analytics_sparse_ticket_dataset] Partition [8] offset [34262356:34262371] (com.clickhouse.kafka.connect.sink.processing.Processing:88)
[2023-06-15 10:13:24,456] INFO [analytics-ticket-clickhouse-sink|task-0]  Number of records to insert 16 to table name analytics_sparse_ticket_dataset (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter:347)
[2023-06-15 10:13:24,513] INFO [analytics-ticket-clickhouse-sink|task-0]  batchSize 16 data ms 0 send 57 (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter:403)
[2023-06-15 10:13:24,517] INFO [analytics-ticket-clickhouse-sink|task-0]  doLogic Topic [analytics_sparse_ticket_dataset] Partition [0] offset [34303211:34303247] (com.clickhouse.kafka.connect.sink.processing.Processing:88)
[2023-06-15 10:13:24,517] INFO [analytics-ticket-clickhouse-sink|task-0]  Number of records to insert 37 to table name analytics_sparse_ticket_dataset (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter:347)
[2023-06-15 10:13:24,570] INFO [analytics-ticket-clickhouse-sink|task-0]  batchSize 37 data ms 0 send 53 (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter:403)
[2023-06-15 10:13:24,724] INFO [analytics-ticket-clickhouse-sink|task-0]  doLogic Topic [analytics_sparse_ticket_dataset] Partition [4] offset [34292593:34292604] (com.clickhouse.kafka.connect.sink.processing.Processing:88)
[2023-06-15 10:13:24,724] INFO [analytics-ticket-clickhouse-sink|task-0]  Number of records to insert 12 to table name analytics_sparse_ticket_dataset (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter:347)
[2023-06-15 10:13:24,790] INFO [analytics-ticket-clickhouse-sink|task-0]  batchSize 12 data ms 0 send 66 (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter:403)
[2023-06-15 10:13:24,792] INFO [analytics-ticket-clickhouse-sink|task-0]  doLogic Topic [analytics_sparse_ticket_dataset] Partition [13] offset [34276191:34276193] (com.clickhouse.kafka.connect.sink.processing.Processing:88)
[2023-06-15 10:13:24,792] INFO [analytics-ticket-clickhouse-sink|task-0]  Number of records to insert 3 to table name analytics_sparse_ticket_dataset (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter:347)
[2023-06-15 10:13:24,856] INFO [analytics-ticket-clickhouse-sink|task-0]  batchSize 3 data ms 0 send 64 (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter:403)

I think that the consumer properties that control this are fetch.max.wait.ms and fetch.min.bytes

However, it doesn't matter how I set them. they don't seem to have any effect.

In connect-distributed.properties, I tried with and without consumer. prefix

...
consumer.fetch.min.bytes=10000000
consumer.fetch.max.wait.ms=10000
fetch.min.bytes=10000000
fetch.max.wait.ms=10000
...

Did anyone succeed at this? I'd love to see a working example 🙏

[mshustov]

@abdelhakimbendjabeur, in what environment do you run the connector?