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
