[KAFKA-12391] Add an option to store arbitrary metadata to a SourceRecord

• Type: Improvement
• Status: Resolved
• Priority: Minor
• Resolution: Information Provided
• Component/s: connect
• Assignee: Unassigned
• Reporter: Luca Burgazzoli
• Created: 01/Mar/21 10:26
• Updated: 24/Mar/21 15:05
• Resolved: 24/Mar/21 15:05

Description

When writing Source Connectors for Kafka, it may be required to perform some additional house cleaning when an record has been acknowledged by the Kafka broker and as today, it is possible to set up an hook by overriding SourceTask.commitRecord(SourceRecord).

This works fine in most of the cases but to make it easy for the source connector to perform it's internal house keeping, it would be nice to have an option to set some additional metadata to the SourceRecord without having impacts to the Record sent to the Kafka Broker, something like:

class SourceRecord {
    public SourceRecord(
        ...,
        Map<String, ?> attributes) {
        ...
        this.attributes = attributes;
    }

    Map<String, ?> attributes() { 
        return attributes;
    }
}