Does batching actually work?

[MarkCWirt]

I am trying to get batching working with the connect client (on a self-hosted, vanilla Kafka deployment), and no matter what I try my batch size is always one record.

The documentation only mentions

consumer.max.poll.records=[NUMBER OF RECORDS]
consumer.max.partition.fetch.bytes=[NUMBER OF RECORDS * RECORD SIZE IN BYTES]

but they don't seem to do anything. I've tried setting them in consumer.properties. I've tried setting them (with consumer.override. in the worker configuration, including connector.client.config.override.policy=All in the consumer properties. I've tries setting consumer.fetch.min.bytes= also. No matter what I've done (and I've been working a full 1 1/2 days on it), I always get a batch size of one.

2023-11-21 14:52:30,694] INFO [audit_log_tuple|task-0] Number of records to insert 1 to table name analytics-audit (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter:547)
[2023-11-21 14:52:30,700] INFO [audit_log_tuple|task-0] batchSize 1 data ms 1 send 5 (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter:637)
[2023-11-21 14:52:30,797] INFO [audit_log_tuple|task-0] doLogic Topic [analytics-audit] Partition [9] offset [20:20] (com.clickhouse.kafka.connect.sink.processing.Processing:87)
[2023-11-21 14:52:30,798] INFO [audit_log_tuple|task-0] Number of records to insert 1 to table name analytics-audit (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter:547)
[2023-11-21 14:52:30,803] INFO [audit_log_tuple|task-0] batchSize 1 data ms 1 send 4 (com.clickhouse.kafka.connect.sink.db.ClickHouseWriter:637)

Does anyone have this working, and if so, can they share a working configuration?

TIA

[Paultagoras]

Hi @MarkCWirt ! So it kind of depends on a few quirks of Kafka Connect:

• Properties are defined differently in a few different places, so even though max.poll.records is a property, it might be set as CONNECT_CONSUMER_MAX_POLL_RECORDS (like in my docker compose file) or consumer.override.max.poll.records (like on Confluent). This also applies to fetch.min.bytes as you can imagine.
• There's another property called fetch.max.wait.ms that can govern when polling happens, so I would make sure you're setting this as well.

The connector itself doesn't do batching/caching beyond the built-in Connect mechanisms, and those properties are what controls it. (A fuller listing could be found here)

Hope that helps!

[MarkCWirt]

Thanks you for you input, @Paultagoras !

I think I made some progress. An issue I was having was knowing what to name the properties. In the consumer.properties file it seems like one doesn't/can't use the consumer. prefix, but in the connect configuration (connect-standalone.properties in my case), one does need to add the prefix. [I'm not only new to Clickhouse: I'm new to Kafka, too.]

So my connect-standalone.properties now has:

consumer.fetch.min.bytes=200000
consumer.fetch.max.wait.ms=60000

plugin.path=/path/to/my/plugins/

and it seems to be working: my records are around 1k, and the connector seems to be batching about 200 records, which would be expected

(I'm not really sure how maximum wait time and minimum bytes interact with one another, but now that batching is taking place I should be able to figue it out.)

Once again, thanks for your help!

[Paultagoras]

That's great to hear! Essentially, the time is the longest time the server will wait before firing off the request - it's the MAX time it will wait. If the size of the data exceeds the set minimum before that point, it'll send off right away.