CODEWITHSUNDEEP
apache-kafkaapache-kafka-connect
user2298491
user2298491

Reputation: 117

Kafka connect consumer referencing offset and storing in message

If I am using kafka-connect to consume messages and store to s3 (using the kafka-connect s3 connector), is there anyway I can store the message offset along with the event payload? I would like to have this data to put some order on the messages and also to check if there could be any gaps or check if there were any duplicates in the messages I have received. (e.g. if my consumer offsets get accidentally clobbered and I restarted kafka-connect). Is this possible or should I write a custom subscriber for this type of functionality?

Upvotes: 3

Views: 956

Answers (1)

Michael Heil
Michael Heil

Reputation: 18525

According to the documentation on Insert Field transformation, you could use offset.field:

Name            Description
offset.field    Field name for Apache Kafka® offset. This is only applicable to sink connectors. Suffix with ! to make this a required field, or ? to keep it optional (the default).

Overall, your single message transformation (SMT) configuration would look like this:

"transforms": "InsertField",
"transforms.InsertField.type": "org.apache.kafka.connect.transforms.InsertField$Value",
"transforms.InsertField.offset.field": "offsetColumn"

If this is not what you are looking for, then there is always the option to create your customised transformations

Upvotes: 3

Related Questions