Kafka message committed before inserted

[Ilanlido]

Hey, i've managed to run the connector locally (not the lightweight one), i've inserted large amount of data to my topic and i noticed that the sink connector committed to the latest offset after short time, meanwhile the amount of rows in clickhouse were incrementing. i then restarted kafka connect and saw that it stopped inserting rows to clickhouse, for example it ended up with 50k rows in clickhouse whereas i originally had 800k messages in the topic. am i missing something on the configuration side?
i would've assumed it would commit the message only after it was actually inserted to clickhouse.
Thanks

[Ilanlido]

I see that there is some logic around handling offsets in acknowledgeRecords
but the if statement "record.getCommitter() != null && record.getSourceRecord() != null" returns false, so i guess this is not the part that actually commits the offsets, also looking at the implementation of preCommit() it seems that it just commits whatever was given, without actually checking what's inflight and what is currently 'in memory' so is this a bug?

[subkanthi]

Hi @Ilanlido , that logic only applies to the lightweight version, the kafka version is in ClickHouseSinkTask.java. Yes currently the automatic kafka offset management is used, so there is no guarantee that the offsets will be commited after records are persisted in ClickHouse, since the records are processed in a separate thread group after they are added to put() in ClickHouseSinkTask.

[Ilanlido]

What could be the reason to implement it only in the lightweight version? can i run the lightweight version as a jar plugin to kafka connect or do i must run it as a separate container? if not, how hard do you think to it will be to implement the logic to gurantee offsets will be committed after records are persisted for the non lightweight version?

[subkanthi]

No the lightweight version does not include any kafka dependencies. The logic is straight forward, the threads in the thread pool need to notify SinkTask when they have persisted to ClickHouse and then the offsets can be acknowledged to Kafka. There are examples in other kafka sink connectors.