...

1. Add new constructor to Source Record

   Code Block
   public class SourceRecord {   + public SourceRecord(Map<String, ?> sourcePartition, Map<String, ?> sourceOffset, String topic, Integer partition, Schema keySchema, Object key, Schema valueSchema, Object value, Long timestamp, Iterable<Header> headers) }

2. Add new constructor to Sink Record

   Code Block
   public class SinkRecord { + public SinkRecord(String topic, int partition, Schema keySchema, Object key, Schema valueSchema, Object value, long kafkaOffset, Long timestamp, TimestampType timestampType, Iterable<Header> headers) }

3. modify parent ConnectRecord constructor , and expose headers accessor method.

   Code Block
   public class SinkRecordConnectRecord {   + public ConnectRecord(String topic, Integer kafkaPartition, Schema keySchema, Object key, Schema valueSchema, Object value, Long timestamp, Iterable<Header> headers) + public Headers headers() }

4. add newRecord methods in Sink and Source to take headers parameter, with existing signature delgating to new signature, but left in for compatbility

   Add newRecord abstract method to ConnectRecord and override implementations in SourceRecord and SinkRecord to take headers parameter, with existing signature delegating to new signature, but left in for compatibility.

Proposed Changes

...