[GitHub] [kafka] cshannon commented on a diff in pull request #12412: KAFKA-14079 - Clear out SubmittedRecords in WorkerSourceTask on failed producer errors

[GitBox <gi...@apache.org>]

cshannon commented on code in PR #12412:
URL: https://github.com/apache/kafka/pull/12412#discussion_r922874851


##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java:
##########
@@ -375,6 +375,10 @@ private boolean sendRecords() {
                                 // executeFailed here allows the use of existing logging infrastructure/configuration
                                 retryWithToleranceOperator.executeFailed(Stage.KAFKA_PRODUCE, WorkerSourceTask.class,
                                         preTransformRecord, e);
+
+                                //Make sure to clear out the record and update metrics
+                                submittedRecords.removeLastOccurrence(submittedRecord);

Review Comment:
   Yeah I think you are right we can just ack instead of removing. It's a failed record so it wouldn't be retried, if should be skipped. I can change that and also take a look at making test changes as well that you commented on.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org