In the class - org.apache.kafka.connect.runtime.WorkerSourceTask
enter function sendRecords()
we know that:

firstly, execute
-
// Offsets are converted & serialized in the OffsetWriter
offsetWriter.offset(record.sourcePartition(), record.sourceOffset());
-
then execute
producer.send(...)

but if producer.send(...) fail, but the offset of the record has been submit to offsetWriter

and we know that,
org.apache.kafka.connect.runtime.SourceTaskOffsetCommitter.schedule(ConnectorTaskId id, WorkerSourceTask workerTask)  is invoked periodically！！！
then it is possible that the offset of fail record has been recroded into the default file : /tmp/connect.offsets

so it is wrong

Is this a bug?