C0urante commented on a change in pull request #10563:
URL: https://github.com/apache/kafka/pull/10563#discussion_r623328064
##########
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'm actually wondering if there's a bug in the existing code; not sure
`pausedForRedelivery` should be reset to `false` after a consumer rebalance. If
the worker is in the middle of retrying a failed batch delivery, all the
partitions will be paused, and after the rebalance, nothing explicitly resumes
them.
During normal execution (when the connector isn't in the middle of
redelivering a failed batch) this doesn't surface because partition consumption
hasn't been paused anyways. But that assumption breaks down if we've paused to
retry a failed batch.
I think it might be necessary to throw in a call to `resumeAll()` if
`pausedForRedelivery` transitions from `true` to `false` as a result of a
consumer rebalance (or, if we're lazy, we can just throw it in unconditionally
at the end of each consumer rebalance if `pausedForRedelivery` is `false` since
[`Consumer::resume` is a no-op for partitions that weren't already
paused](https://kafka.apache.org/27/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#resume-java.util.Collection-)).
--
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
