C0urante commented on a change in pull request #10563: URL: https://github.com/apache/kafka/pull/10563#discussion_r623321728
########## 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: > Setting this to anything other than false causes us not to resume partitions which we own that were not explicitly requested to be paused. Yep, that's intentional. The retry loop for sink tasks works like this: - Grab new batch of records from consumer, store in `messageBatch` - Deliver batch to task via `SinkTask::put` - Catch a `RetriableException` from the task - Pause all partitions on the consumer and set `pausedForRedelivery` to `true` - Redeliver batch (stored in the `messageBatch` field) to task via `SinkTask::put` continuously - Once a delivery attempt is successful, clear the batch stored in `messageBatch` and, since `pausedForRedelivery` is `true`, resume all partitions on the consumer and reset `pausedForDelivery` to `false` When consumer rebalances meant a mass-revocation and mass-reassignment of partitions, we could get away with unconditionally resetting `pausedForRedelivery` to `true` during them, since offsets for any message batches going through the retry loop would not be committed and those records would be guaranteed to be redelivered after the rebalance. However, with incremental consumer rebalances, a task might not be forced to commit offsets for some (or even all) records in its current batch (if, for example, the new assignment only included additional partitions and no revocations). In that case, we have to be careful to remain paused so that the current batch can continue to be retried. -- 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