[KAFKA-15145] AbstractWorkerSourceTask re-processes records filtered out by SMTs on retriable exceptions

• Type: Bug
• Status: Resolved
• Priority: Minor
• Resolution: Fixed
• Affects Version/s: 0.10.2.0, 0.10.2.1, 0.10.2.2, 0.11.0.0, 0.11.0.1, 0.11.0.2, 0.11.0.3, 1.0.0, 1.0.1, 1.0.2, 1.1.0, 1.1.1, 2.0.0, 2.0.1, 2.1.0, 2.2.0, 2.1.1, 2.3.0, 2.2.1, 2.2.2, 2.4.0, 2.3.1, 2.5.0, 2.4.1, 2.6.0, 2.5.1, 2.7.0, 2.6.1, 2.8.0, 2.7.1, 2.6.2, 3.1.0, 2.6.3, 2.7.2, 2.8.1, 3.0.0, 3.0.1, 2.8.2, 3.2.0, 3.1.1, 3.3.0, 3.0.2, 3.1.2, 3.2.1, 3.4.0, 3.2.2, 3.2.3, 3.3.1, 3.3.2, 3.5.0, 3.4.1
• Fix Version/s: 3.6.0, 3.5.1
• Component/s: connect
• Assignee: Yash Mayya
• Reporter: Yash Mayya
• Created: 04/Jul/23 10:44
• Updated: 10/Jul/23 17:34
• Resolved: 10/Jul/23 17:27

Description

If a RetriableException is thrown from an admin client or producer client operation in AbstractWorkerSourceTask::sendRecords, the send operation is retried for the remaining records in the batch. There is a bug in the logic for computing the remaining records in a batch which causes records that are filtered out by the task's transformation chain to be re-processed. This will also result in the SourceTask::commitRecord method being called twice for the same record, which can cause certain types of source connectors to fail. This bug seems to exist since when SMTs were first introduced in 0.10.2

Issue Links

• links to: GitHub Pull Request #13955