C0urante commented on a change in pull request #10563: URL: https://github.com/apache/kafka/pull/10563#discussion_r623409533
########## File path: connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java ########## @@ -680,13 +717,13 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) { } sinkTaskMetricsGroup.assignedOffsets(currentOffsets); - // If we paused everything for redelivery (which is no longer relevant since we discarded the data), make + // If we paused everything for redelivery and all partitions for the failed deliveries have been revoked, make // sure anything we paused that the task didn't request to be paused *and* which we still own is resumed. // Also make sure our tracking of paused partitions is updated to remove any partitions we no longer own. - pausedForRedelivery = false; + pausedForRedelivery = pausedForRedelivery && !messageBatch.isEmpty(); Review comment: Oooooh, one more edge case to consider--if new partitions are assigned while in the middle of retrying a failed batch, right now the worker might accidentally receive new records from those partitions, which would violate the assertion that new records can be received from the consumer if and only if the `messageBatch` field is empty. I think we can (and should) handle this by invoking `pauseAll()` if `pausedForRedelivery` is still `true` after removing records from revoked partitions from the current batch. -- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org