Skip to content

KAFKA-12487: Add support for cooperative consumer protocol with sink connectors #10563

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

Merged
kkonstantine merged 6 commits into from
Nov 10, 2021
Merged

KAFKA-12487: Add support for cooperative consumer protocol with sink connectors #10563

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
7708335
KAFKA-12487: Add support for cooperative consumer protocol with sink …
C0urante Apr 6, 2021
05aa4f6
KAFKA-12487: Fix sink task consumer pause/resume logic while paused f…
C0urante Apr 29, 2021
ec1c6db
KAFKA-12487: Fix unit test
C0urante May 12, 2021
6289b22
KAFKA-12487: Fix errant record reporter logic for multiple records wi…
C0urante Jul 16, 2021
9be00f6
KAFKA-12487: Address review feedback
C0urante Jul 16, 2021
db1f8f5
KAFKA-12487: Test cleanup
C0urante Jul 16, 2021
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
169 changes: 117 additions & 52 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 @@ -61,6 +61,7 @@
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import static java.util.Collections.singleton;
import static org.apache.kafka.connect.runtime.WorkerConfig.TOPIC_TRACKING_ENABLE_CONFIG;
Expand Down Expand Up @@ -125,6 +126,7 @@ public WorkerSinkTask(ConnectorTaskId id,
this.headerConverter = headerConverter;
this.transformationChain = transformationChain;
this.messageBatch = new ArrayList<>();
this.lastCommittedOffsets = new HashMap<>();
this.currentOffsets = new HashMap<>();
this.origOffsets = new HashMap<>();
this.pausedForRedelivery = false;
Expand Down Expand Up @@ -196,7 +198,7 @@ public void execute() {
initializeAndStart();
// Make sure any uncommitted data has been committed and the task has
// a chance to clean up its state
try (UncheckedCloseable suppressible = this::closePartitions) {
try (UncheckedCloseable suppressible = this::closeAllPartitions) {
while (!isStopping())
iteration();
} catch (WakeupException e) {
Expand Down Expand Up @@ -367,42 +369,54 @@ private void doCommit(Map<TopicPartition, OffsetAndMetadata> offsets, boolean cl
}

private void commitOffsets(long now, boolean closing) {
commitOffsets(now, closing, consumer.assignment());
}

private void commitOffsets(long now, boolean closing, Collection<TopicPartition> topicPartitions) {
log.trace("Committing offsets for partitions {}", topicPartitions);
if (workerErrantRecordReporter != null) {
log.trace("Awaiting all reported errors to be completed");
workerErrantRecordReporter.awaitAllFutures();
log.trace("Completed all reported errors");
log.trace("Awaiting reported errors to be completed");
workerErrantRecordReporter.awaitFutures(topicPartitions);
log.trace("Completed reported errors");
}

if (currentOffsets.isEmpty())
Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = currentOffsets.entrySet().stream()
.filter(e -> topicPartitions.contains(e.getKey()))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

if (offsetsToCommit.isEmpty())
return;

committing = true;
commitSeqno += 1;
commitStarted = now;
sinkTaskMetricsGroup.recordOffsetSequenceNumber(commitSeqno);

Map<TopicPartition, OffsetAndMetadata> lastCommittedOffsetsForPartitions = this.lastCommittedOffsets.entrySet().stream()
.filter(e -> offsetsToCommit.containsKey(e.getKey()))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

final Map<TopicPartition, OffsetAndMetadata> taskProvidedOffsets;
try {
log.trace("{} Calling task.preCommit with current offsets: {}", this, currentOffsets);
taskProvidedOffsets = task.preCommit(new HashMap<>(currentOffsets));
log.trace("{} Calling task.preCommit with current offsets: {}", this, offsetsToCommit);
taskProvidedOffsets = task.preCommit(new HashMap<>(offsetsToCommit));
} catch (Throwable t) {
if (closing) {
log.warn("{} Offset commit failed during close", this);
onCommitCompleted(t, commitSeqno, null);
} else {
log.error("{} Offset commit failed, rewinding to last committed offsets", this, t);
for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : lastCommittedOffsets.entrySet()) {
for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : lastCommittedOffsetsForPartitions.entrySet()) {
log.debug("{} Rewinding topic partition {} to offset {}", this, entry.getKey(), entry.getValue().offset());
consumer.seek(entry.getKey(), entry.getValue().offset());
}
currentOffsets = new HashMap<>(lastCommittedOffsets);

@kkonstantine kkonstantine Jul 14, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this change means we never create a fresh copy of the offsets map.
Is this correct? Is there a risk for the offsets to keep being added and never removed?

@C0urante C0urante Jul 16, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch. I believe this would lead to incorrect metrics given the use of the currentOffsets field when invoking sinkTaskMetricsGroup::assignedOffsets, and may lead to issues with the end-of-life call to closePartitions as well.

I'll add cleanup logic in the consumer rebalance listener.

onCommitCompleted(t, commitSeqno, null);
currentOffsets.putAll(lastCommittedOffsetsForPartitions);
}
onCommitCompleted(t, commitSeqno, null);
return;
} finally {
if (closing) {
log.trace("{} Closing the task before committing the offsets: {}", this, currentOffsets);
task.close(currentOffsets.keySet());
log.trace("{} Closing the task before committing the offsets: {}", this, offsetsToCommit);
task.close(topicPartitions);
}
}

Expand All @@ -412,32 +426,36 @@ private void commitOffsets(long now, boolean closing) {
return;
}

final Map<TopicPartition, OffsetAndMetadata> commitableOffsets = new HashMap<>(lastCommittedOffsets);
Collection<TopicPartition> allAssignedTopicPartitions = consumer.assignment();
final Map<TopicPartition, OffsetAndMetadata> committableOffsets = new HashMap<>(lastCommittedOffsetsForPartitions);
for (Map.Entry<TopicPartition, OffsetAndMetadata> taskProvidedOffsetEntry : taskProvidedOffsets.entrySet()) {
final TopicPartition partition = taskProvidedOffsetEntry.getKey();
final OffsetAndMetadata taskProvidedOffset = taskProvidedOffsetEntry.getValue();
if (commitableOffsets.containsKey(partition)) {
if (committableOffsets.containsKey(partition)) {
long taskOffset = taskProvidedOffset.offset();
long currentOffset = currentOffsets.get(partition).offset();
long currentOffset = offsetsToCommit.get(partition).offset();
if (taskOffset <= currentOffset) {
commitableOffsets.put(partition, taskProvidedOffset);
committableOffsets.put(partition, taskProvidedOffset);
} else {
log.warn("{} Ignoring invalid task provided offset {}/{} -- not yet consumed, taskOffset={} currentOffset={}",
this, partition, taskProvidedOffset, taskOffset, currentOffset);
this, partition, taskProvidedOffset, taskOffset, currentOffset);
}
} else {
} else if (!allAssignedTopicPartitions.contains(partition)) {
log.warn("{} Ignoring invalid task provided offset {}/{} -- partition not assigned, assignment={}",
this, partition, taskProvidedOffset, consumer.assignment());
this, partition, taskProvidedOffset, allAssignedTopicPartitions);
} else {
log.debug("{} Ignoring task provided offset {}/{} -- partition not requested, requested={}",
this, partition, taskProvidedOffset, committableOffsets.keySet());
}
}

if (commitableOffsets.equals(lastCommittedOffsets)) {
if (committableOffsets.equals(lastCommittedOffsetsForPartitions)) {
log.debug("{} Skipping offset commit, no change since last commit", this);
onCommitCompleted(null, commitSeqno, null);
return;
}

doCommit(commitableOffsets, closing, commitSeqno);
doCommit(committableOffsets, closing, commitSeqno);
}


Expand Down Expand Up @@ -598,10 +616,12 @@ private void deliverMessages() {
}
} catch (RetriableException e) {
log.error("{} RetriableException from SinkTask:", this, 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;
pauseAll();
if (!pausedForRedelivery) {
// 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;
pauseAll();
}
// Let this exit normally, the batch will be reprocessed on the next loop.
} catch (Throwable t) {
log.error("{} Task threw an uncaught and unrecoverable exception. Task is being killed and will not "
Expand Down Expand Up @@ -631,13 +651,32 @@ private void rewind() {
}

private void openPartitions(Collection<TopicPartition> partitions) {
sinkTaskMetricsGroup.recordPartitionCount(partitions.size());
updatePartitionCount();
task.open(partitions);
}

private void closePartitions() {
commitOffsets(time.milliseconds(), true);
sinkTaskMetricsGroup.recordPartitionCount(0);
private void closeAllPartitions() {
closePartitions(currentOffsets.keySet(), false);
}

private void closePartitions(Collection<TopicPartition> topicPartitions, boolean lost) {
if (!lost) {
commitOffsets(time.milliseconds(), true, topicPartitions);
} else {
log.trace("{} Closing the task as partitions have been lost: {}", this, topicPartitions);
task.close(topicPartitions);
if (workerErrantRecordReporter != null) {
log.trace("Cancelling reported errors for {}", topicPartitions);
workerErrantRecordReporter.cancelFutures(topicPartitions);

@ncliang ncliang Apr 29, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure if cancelling the outstanding futures for error reporting is the right thing to do here. Would it be reasonable to await their completion for a reasonable amount of time before giving up?

@C0urante C0urante Apr 29, 2021 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In this case, we've lost the partition assignment and won't be able to commit offsets for them, and since offset commits don't take place before outstanding error reports have completed, we know that no offsets for the records that caused these error reports have been committed either. So, we're guaranteed that the records that the task reported errors for will be redelivered, to whichever task now owns these partitions, regardless of whether we wait here for the reporting to complete.

@ncliang ncliang May 1, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see. So the distinction is because in the revoked case we get a chance to await on the errant record reporter futures before we commit offsets for the revoked partitions, but in the lost case we've already lost the partition and don't get a chance to commit offsets. Since we already do not own the partition, we should not be reporting errors for it and should let the current owner take that responsibility. It should be noted that the cancelation is best effort, so there is a chance we duplicate reporting for the errant record.

log.trace("Cancelled all reported errors for {}", topicPartitions);
}
topicPartitions.forEach(currentOffsets::remove);
}
updatePartitionCount();
}

private void updatePartitionCount() {
sinkTaskMetricsGroup.recordPartitionCount(consumer.assignment().size());
}

@Override
Expand Down Expand Up @@ -670,8 +709,7 @@ private class HandleRebalance implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.debug("{} Partitions assigned {}", WorkerSinkTask.this, partitions);
lastCommittedOffsets = new HashMap<>();

@kkonstantine kkonstantine Jul 14, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

similar question as above. This is a map that only grows now. Is this correct?

@C0urante C0urante Jul 16, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(Addressed above; it is almost certainly not correct)

currentOffsets = new HashMap<>();

for (TopicPartition tp : partitions) {
long pos = consumer.position(tp);
lastCommittedOffsets.put(tp, new OffsetAndMetadata(pos));
Expand All @@ -680,17 +718,29 @@ 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
// 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;

// Ensure that the paused partitions contains only assigned partitions and repause as necessary
context.pausedPartitions().retainAll(partitions);
if (shouldPause())
boolean wasPausedForRedelivery = pausedForRedelivery;
pausedForRedelivery = wasPausedForRedelivery && !messageBatch.isEmpty();
if (pausedForRedelivery) {
// Re-pause here in case we picked up new partitions in the rebalance
pauseAll();
else if (!context.pausedPartitions().isEmpty())
consumer.pause(context.pausedPartitions());
} else {
// 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.
if (wasPausedForRedelivery) {
resumeAll();
}
// Ensure that the paused partitions contains only assigned partitions and repause as necessary
context.pausedPartitions().retainAll(consumer.assignment());
if (shouldPause())
pauseAll();
else if (!context.pausedPartitions().isEmpty())
consumer.pause(context.pausedPartitions());
}

if (partitions.isEmpty()) {
return;
}

// Instead of invoking the assignment callback on initialization, we guarantee the consumer is ready upon
// task start. Since this callback gets invoked during that initial setup before we've started the task, we
Expand All @@ -710,22 +760,35 @@ else if (!context.pausedPartitions().isEmpty())

@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
onPartitionsRemoved(partitions, false);
}

@Override
public void onPartitionsLost(Collection<TopicPartition> partitions) {
onPartitionsRemoved(partitions, true);
}

private void onPartitionsRemoved(Collection<TopicPartition> partitions, boolean lost) {
if (taskStopped) {
log.trace("Skipping partition revocation callback as task has already been stopped");
return;
}
log.debug("{} Partitions revoked", WorkerSinkTask.this);
log.debug("{} Partitions {}: {}", WorkerSinkTask.this, lost ? "lost" : "revoked", partitions);

if (partitions.isEmpty())
return;

try {
closePartitions();
sinkTaskMetricsGroup.clearOffsets();
closePartitions(partitions, lost);
sinkTaskMetricsGroup.clearOffsets(partitions);
} catch (RuntimeException e) {
// The consumer swallows exceptions raised in the rebalance listener, so we need to store
// exceptions and rethrow when poll() returns.
rebalanceException = e;
}

// Make sure we don't have any leftover data since offsets will be reset to committed positions
messageBatch.clear();
// Make sure we don't have any leftover data since offsets for these partitions will be reset to committed positions
messageBatch.removeIf(record -> partitions.contains(new TopicPartition(record.topic(), record.kafkaPartition())));

@kkonstantine kkonstantine Jul 14, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do you worry that this became more expensive now, especially in cases where we want to remove everything as before?

@C0urante C0urante Jul 16, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It certainly will be more expensive, although this will only occur in an edge case where the task has thrown a RetriableException from SinkTask::put, then before a follow-up invocation of SinkTask::put is able to succeed, a consumer rebalance takes place and partitions are revoked.

A naive improvement might be to convert messageBatch to a Map<TopicPartition, List<SinkRecord>>, which would allow us to quickly filter out records belonging to a given topic partition. But that would also be less efficient since we'd have to re-join those lists together before delivering records to the task, and would have to populate that map in the first place after retrieving the original list of records from Consumer::poll.

Not sure if there's a good way around this; if you believe it's a blocker I'm happy to spend some more time on it, though.

}
}

Expand Down Expand Up @@ -844,13 +907,15 @@ void recordCommittedOffsets(Map<TopicPartition, OffsetAndMetadata> offsets) {
void assignedOffsets(Map<TopicPartition, OffsetAndMetadata> offsets) {
consumedOffsets = new HashMap<>(offsets);
committedOffsets = offsets;
sinkRecordActiveCount.record(0.0);
computeSinkRecordLag();
}

void clearOffsets() {
consumedOffsets.clear();
committedOffsets.clear();
sinkRecordActiveCount.record(0.0);
void clearOffsets(Collection<TopicPartition> topicPartitions) {
topicPartitions.forEach(tp -> {
consumedOffsets.remove(tp);
committedOffsets.remove(tp);
});
computeSinkRecordLag();
}

void recordOffsetCommitSuccess() {
Expand Down
Loading