...

public abstract class SourceTask implements Task {
---%<---


// existing method
public void commitRecord(SourceRecord sourceRecord) {
  // nop
}


// new method calls old one by default
public void commitRecord(SourceRecord sourceRecord, RecordMetadata recordMetadata) {
  // nopcommitRecord(sourceRecord);
}
---%<---

Proposed Changes

...

The new overloaded version will be invoked only when a record is ACK'd, which implies the record was not filtered and was not skipped. This is somewhat intuitive, as the RecordMetadata can only be filled in by a producer ACK.

After an ACK, WorkerSourceTask currently logs MetadataRecord.offset() and partition() before invoking the commitRecord() callback. I propose to also invoke the new overloaded version as well:

task.commitRecord(preTransformRecord);

if (recordMetadata != null)
    task.commitRecord(preTransformRecord, recordMetadata);

instead, with recordMetadata null when there is no ACK. To preserve backwards compatibility, the default implementation will call the old method.

Compatibility, Deprecation, and Migration Plan

This is a new callback and won't affect existing code beyond an additional no-op function call. 

Rejected Alternatives

• We could introduce a new method name, but overloading an existing method is a lighter touch.
• We could invoke commitRecord() with recordMetadata == null when records are skipped, but this would be surprising and error prone to those using the APIboth old and new methods, but this yields a confusing interface, where it is unclear which method to implement.
• SinkRecord has offset and partition fields, so we could potentially use commitRecord(SinkRecord), but this would be a confusing abuse of SinkRecord's semantics. In particular, it would be weird to have SinkRecord in the SourceTask interface.
• We could add kafkaOffset and kafkaPartition fields to SourceRecord, which would be null prior to ACK and then filled in by WorkerSourceTask after ACK. However, this breaks the "value class" semantics of SourceRecord.
• We could extend SourceRecord with AckedSourceRecord or LoggedSourceRecord, but this seems overkill.
• We could avoid using RecordMetadata in the API, and instead include kafkaOffset, kafkaPartition, kafkaTopic, and kafkaTimestamp as parameters to commitRecord(). But this is a lot of parameters, and RecordMatadata is already in the clients API.
• We could pass just the kafkaOffset to commitRecord(), but an offset is mostly meaningless without an associated partition. We could include both kafkaOffset and kafkaPartition, but this doesn't account for transformations that may change the downstream topic name. We could include kafkaTopic as well, but then we might as well include the entire RecordMetadata.