C0urante commented on a change in pull request #10563:
URL: https://github.com/apache/kafka/pull/10563#discussion_r623442240
##########
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:
I've pushed a change that should address this gap and includes a new
unit test for the pause/resume logic while paused for redelivery across
consumer rebalances.
--
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
