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
