[KAFKA-8540] Improve InsertField connect transform to allow inserting processed timestamp

• Type: Improvement
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 2.2.1
• Component/s: connect
• Labels: needs-kip
• Flags: Important
• Assignee: Unassigned
• Reporter: Khushal Kumthekar
• Created: 14/Jun/19 17:27
• Updated: 22/Jul/19 15:10

Description

We have requirement to insert a new field that gets populated with the timestamp when kafka message was processed by the connector.

There is no such transform currently available. "InputField" transform provides timestamp.field option but that sets the field value to the timestamp when messaged landed in kafka and not when message was processed by the connector.

Here is the proposal

Enhance InputField transform to add a new option called "processedTimestamp.field" that will allow users to insert new field with value of message processed timestamp as shown in below configuration

"transforms": "InsertField",
"transforms.InsertField.type":"org.apache.kafka.connect.transforms.InsertField$Value",
"transforms.InsertField.processedTimestamp.field": "field_name"

here is the potential code block to set this value

updatedValue.put(processedTimestampField.name, new Date());