Skip to content

KAFKA-2786: Only respond to SinkTask onPartitionsRevoked after the WorkerSinkTask has finished starting up. #476

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ewencp wants to merge 1 commit into from
Closed

KAFKA-2786: Only respond to SinkTask onPartitionsRevoked after the WorkerSinkTask has finished starting up. #476

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 6 additions & 4 deletions connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,7 +191,7 @@ public void commitOffsets(boolean sync, final int seqno) {
try {
task.flush(offsets);
} catch (Throwable t) {
log.error("Commit of {} offsets failed due to exception while flushing: {}", this, t);
log.error("Commit of {} offsets failed due to exception while flushing:", this, t);
log.error("Rewinding offsets to last committed offsets");
for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : lastCommittedOffsets.entrySet()) {
log.debug("{} Rewinding topic partition {} to offset {}", id, entry.getKey(), entry.getValue().offset());
Expand Down Expand Up @@ -288,7 +288,7 @@ private void deliverMessages() {
pausedForRedelivery = false;
}
} catch (RetriableException e) {
log.error("RetriableException from SinkTask {}: {}", id, e);
log.error("RetriableException from SinkTask {}:", id, e);
// If we're retrying a previous batch, make sure we've paused all topic partitions so we don't get new data,
// but will still be able to poll in order to handle user-requested timeouts, keep group membership, etc.
pausedForRedelivery = true;
Expand Down Expand Up @@ -361,8 +361,10 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {

@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
task.onPartitionsRevoked(partitions);
commitOffsets(true, -1);
if (started) {
task.onPartitionsRevoked(partitions);
commitOffsets(true, -1);
}
// Make sure we don't have any leftover data since offsets will be reset to committed positions
messageBatch.clear();
}
Expand Down