Issue : fetch.max.wait.ms setting of kafka connect not working as expected with CH sink connector

[rapsvenkat]

Broker is running with below configuration

socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
fetch.max.bytes=419430400
replica.socket.receive.buffer.bytes=131072
num.replica.fetchers=2
message.max.bytes=2097176
replica.fetch.max.bytes=2097176

The broker is configured to return up to 400 MB for a single fetch request.

Kafka Connect ClickHouse sink connector running with the configuration below
consumer.override.group.id: test.1",
connector.class: "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
consumer.override.max.poll.records: 400000,
consumer.override.fetch.max.wait.ms: 60000,
consumer.override.max.partition.fetch.bytes": 109715200,
tasks.max: "8",
topics: "sna.heartbeat",

Our goal is to fetch and poll more records, then insert a large batch of records into ClickHouse.

[2024-06-03 13:04:06,045] INFO doLogic - Topic: [sna.heartbeat], Partition: [5], MinOffset: [8524935], MaxOffset: [8529101], Records: [4167] (com.clickhouse.kafka.connect.sink.processing.Processing)
[2024-06-03 13:04:06,142] INFO doLogic - Topic: [sna.heartbeat], Partition: [5], MinOffset: [8529102], MaxOffset: [8529130], Records: [29] (com.clickhouse.kafka.connect.sink.processing.Processing)
 [2024-06-03 13:04:06,264] INFO doLogic - Topic: [sna.heartbeat], Partition: [5], MinOffset: [8529131], MaxOffset: [8536021], Records: [6891] (com.clickhouse.kafka.connect.sink.processing.Processing)

But according to the above log, the CH Sink connector is inserting for very few records and making calls to ClickHouse every second.

But if we update the connector config to include the parameter consumer.override.fetch.min.bytes: 109715200, then the CH Sink connector waits and inserts 400k records with every push. We have the following questions:

1. Do we need to set both fetch.max.wait.ms and fetch.min.bytes to insert large batches effectively, or is fetch.max.wait.ms alone sufficient?
2. Is there any setting to ensure that the CH Sink connector inserts records every minute while following the fetch.max.wait.ms setting without specifying fetch.min.bytes?

[Paultagoras]

Hi @rapsvenkat!

So fetch.max.wait.ms works in conjunction with fetch.min.bytes

Kafka Connect will keep waiting for the minimum bytes you specify (fetch.min.bytes) until the timeout is reached (fetch.max.wait.ms) - if you wanted it to be a more regular cadence, you could try setting fetch.min.bytes to some comically high value (then it would always trigger via timeout)

If you ONLY set fetch.min.bytes it would still trigger every 500ms (the default value of fetch.max.wait.ms), and if you ONLY set fetch.max.wait.ms it would trigger as soon as data shows up (the default value of fetch.min.bytes is 1 byte).

This is all out-of-the-box functionality of Kafka Connect, so we don't have special settings around batch sizes 🙂

Hope that helps!