[KAFKA-16069] Source Tasks re-transform records after Retriable exceptions

• Type: Bug
• Status: Open
• Priority: Minor
• Resolution: Unresolved
• Component/s: connect
• Assignee: Sudesh Wasnik
• Reporter: Greg Harris
• Created: 29/Dec/23 23:26
• Updated: 14/Feb/24 19:12

Description

In the SinkTask, records which fail to be delivered to the task#put with a Retriable exception are re-delivered on the next iteration. The SourceTask follows a similar pattern, where records which fail to be delivered to Producer#send with a Retriable exception are retried.

However, the SinkTask accumulates the post-transform records, and does not recompute the transformations over again after a retriable exception. The SourceTask does not accumulate ProducerRecords, and instead recomputes the transformations and converters starting from the pre-transformation AbstractWorkerSourceTask#toSend list.

This means that stateful transformations and converters may see the records rewind, without any indication that the records are the same. For stateless transformations and converters, this means that redundant computation is performed that may be better allocated to other tasks.

Issue Links

• relates to: KAFKA-15912 Parallelize conversion and transformation steps in Connect
• links to: GitHub Pull Request #15374