Skip to content

KAFKA-12226: Prevent source task offset failure when producer is overwhelmed #10112

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
C0urante wants to merge 4 commits into from
Closed

KAFKA-12226: Prevent source task offset failure when producer is overwhelmed #10112

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
cc59865
KAFKA-12226: Prevent source task offset failure when producer is over…
C0urante Feb 11, 2021
549fdfb
KAFKA-12226: Address review comments by deduplicating test code
C0urante Feb 11, 2021
03c5a83
KAFKA-12226: Fix assertion in unit test
C0urante Feb 18, 2021
f13033f
KAFKA-12226: Add final modifier to stop request latch field
C0urante Feb 25, 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
2 changes: 1 addition & 1 deletion checkstyle/suppressions.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,7 @@
files="(KafkaConfigBackingStore|Values).java"/>

<suppress checks="NPathComplexity"
files="(DistributedHerder|RestClient|RestServer|JsonConverter|KafkaConfigBackingStore|FileStreamSourceTask|TopicAdmin).java"/>
files="(DistributedHerder|RestClient|RestServer|JsonConverter|KafkaConfigBackingStore|FileStreamSourceTask|TopicAdmin|WorkerSourceTask).java"/>

<!-- connect tests-->
<suppress checks="ClassDataAbstractionCoupling"
Expand Down
32 changes: 19 additions & 13 deletions connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,8 +98,9 @@ class WorkerSourceTask extends WorkerTask {
private IdentityHashMap<ProducerRecord<byte[], byte[]>, ProducerRecord<byte[], byte[]>> outstandingMessages;
// A second buffer is used while an offset flush is running
private IdentityHashMap<ProducerRecord<byte[], byte[]>, ProducerRecord<byte[], byte[]>> outstandingMessagesBacklog;
private boolean flushing;
private CountDownLatch stopRequestedLatch;
private boolean recordFlushPending;
private boolean offsetFlushPending;
private final CountDownLatch stopRequestedLatch;

private Map<String, String> taskConfig;
private boolean started = false;
Expand Down Expand Up @@ -144,7 +145,7 @@ public WorkerSourceTask(ConnectorTaskId id,
this.lastSendFailed = false;
this.outstandingMessages = new IdentityHashMap<>();
this.outstandingMessagesBacklog = new IdentityHashMap<>();
this.flushing = false;
this.recordFlushPending = false;
this.stopRequestedLatch = new CountDownLatch(1);
this.sourceTaskMetricsGroup = new SourceTaskMetricsGroup(id, connectMetrics);
this.producerSendException = new AtomicReference<>();
Expand Down Expand Up @@ -335,7 +336,7 @@ private boolean sendRecords() {
// messages and update the offsets.
synchronized (this) {
if (!lastSendFailed) {
if (!flushing) {
if (!recordFlushPending) {
outstandingMessages.put(producerRecord, producerRecord);
} else {
outstandingMessagesBacklog.put(producerRecord, producerRecord);
Expand Down Expand Up @@ -453,12 +454,12 @@ private void commitTaskRecord(SourceRecord record, RecordMetadata metadata) {
private synchronized void recordSent(final ProducerRecord<byte[], byte[]> record) {
ProducerRecord<byte[], byte[]> removed = outstandingMessages.remove(record);
// While flushing, we may also see callbacks for items in the backlog
if (removed == null && flushing)
if (removed == null && recordFlushPending)
removed = outstandingMessagesBacklog.remove(record);
// But if neither one had it, something is very wrong
if (removed == null) {
log.error("{} CRITICAL Saw callback for record that was not present in the outstanding message set: {}", this, record);
} else if (flushing && outstandingMessages.isEmpty()) {
} else if (recordFlushPending && outstandingMessages.isEmpty()) {
// flush thread may be waiting on the outstanding messages to clear
this.notifyAll();
}
Expand All @@ -475,11 +476,15 @@ public boolean commitOffsets() {
synchronized (this) {
// First we need to make sure we snapshot everything in exactly the current state. This
// means both the current set of messages we're still waiting to finish, stored in this
// class, which setting flushing = true will handle by storing any new values into a new
// class, which setting recordFlushPending = true will handle by storing any new values into a new
// buffer; and the current set of user-specified offsets, stored in the
// OffsetStorageWriter, for which we can use beginFlush() to initiate the snapshot.
flushing = true;
boolean flushStarted = offsetWriter.beginFlush();
// No need to begin a new offset flush if we timed out waiting for records to be flushed to
// Kafka in a prior attempt.
if (!recordFlushPending) {

@hachikuji hachikuji Feb 24, 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.

If I understand it correctly, the main difference in this patch is that we no longer fail the flush if the messages cannot be drained quickly enough from outstandingMessages. A few questions come to mind:

  1. Is the flush timeout still a useful configuration? Was it ever? Even if we timeout, we still have to wait for the records that were sent to the producer.
  2. While we are waiting for outstandingMessages to be drained, we are still accumulating messages in outstandingMessagesBacklog. I imagine we can get into a pattern here once we fill up the accumulator. While we're waiting for outstandingMessages to complete, we fill outstandingMessagesBacklog. Once the flush completes, outstandingMessagesBacklog becomes outstandingMessages and we are stuck waiting again. Could this prevent us from satisfying the commit interval?

Overall, I can't shake the feeling that this logic is more complicated than necessary. Why do we need the concept of flushing at all? It would be more intuitive to just commit whatever the latest offsets are. Note that we do not use outstandingMessages for the purpose of retries. Once a request has been handed off to the producer successfully, we rely on the producer to handle retries. Any delivery failure after that is treated as fatal. So then does oustandingMessages serve any other purpose other than tracking flushing? I am probably missing something here. It has been a long time since I reviewed this logic.

@C0urante C0urante Feb 24, 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.

  1. I think it's a necessary evil, since source task offset commits are conducted on a single thread. Without a timeout for offset commits, a single task could block indefinitely and disable offset commits for all other tasks on the cluster.
  2. This is definitely possible; I think the only saving grace here is that the combined sizes of the outstandingMessages and outstandingMessagesBacklog fields is going to be naturally throttled by the producer's buffer. If too many records are accumulated, the call to Producer::send will block synchronously until space is freed up, at which point, the worker can continue polling the task for new records. This isn't ideal as it will essentially cause the producer's entire buffer to be occupied until the throughput of record production from the task decreases and/or the write throughput of the producer rises to meet it, but it at least establishes an upper bound for how large a single batch of records in the oustandingMessages field ever gets. It may take several offset commit attempts for all of the records in that batch to be ack'd, with all but the last (successful) attempt timing out and failing, but forward progress with offset commits should still be possible.

I share your feelings about the complexity here. I think ultimately it arises from two constraints:

  1. A worker-global producer is used to write source offsets to the internal offsets topic right now. Although this doesn't necessarily require the single-threaded logic for offset commits mentioned above, things become simpler with it.
  2. (Please correct me if I'm wrong on this point; my core knowledge is a little fuzzy and maybe there are stronger guarantees than I'm aware of) Out-of-order acknowledgment of records makes tracking the latest offset for a given source partition a little less trivial than it seems initially. For example, if a task produces two records with the same source partition that end up being delivered to different topic-partitions, the second record may be ack'd before the first, and when it comes time for offset commit, the framework would have to refrain from committing offsets for that second record until the first is also ack'd.

I don't think either of these points make it impossible to add even more-fine-grained offset commit behavior and/or remove offset commit timeouts, but the work involved would be a fair amount heavier than this relatively-minor patch. If you'd prefer to see something along those lines, could we consider merging this patch for the moment and perform a more serious overhaul of the source task offset commit logic as a follow-up, possibly with a small design discussion on a Jira ticket to make sure there's alignment on the new behavior?

@hachikuji hachikuji Feb 24, 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.

(Please correct me if I'm wrong on this point; my core knowledge is a little fuzzy and maybe there are stronger guarantees than I'm aware of) Out-of-order acknowledgment of records makes tracking the latest offset for a given source partition a little less trivial than it seems initially. For example, if a task produces two records with the same source partition that end up being delivered to different topic-partitions, the second record may be ack'd before the first, and when it comes time for offset commit, the framework would have to refrain from committing offsets for that second record until the first is also ack'd.

Ok, that rings a bell. I think I see how the logic works now and I don't see an obvious way to make it simpler. Doing something finer-grained as you said might be the way to go. Anyway, I agree this is something to save for a follow-up improvement.

I think it's a necessary evil, since source task offset commits are conducted on a single thread. Without a timeout for offset commits, a single task could block indefinitely and disable offset commits for all other tasks on the cluster.

Hmm.. This is suspicious. Why do we need to block the executor while we wait for the flush? Would it be simpler to let the worker source task finish the flush and the offset commit in its own event thread? We end up blocking the event thread anyway because of the need to do it under the lock.

@C0urante C0urante Feb 25, 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.

We end up blocking the event thread anyway because of the need to do it under the lock.

I think we actually keep polling the task for records during the offset commit, which is the entire reason we have the outstandingMessagesBacklog field. Without it, we'd just add everything to outstandingMessages knowing that, if we've made it to the point of adding a record to that collection, we're not in the process of committing offset, right?

Concretely, we can see that the offset thread relinquishes the lock on the WorkerSourceTask instance while waiting for outstanding messages to be ack'd.

I'm not sure we need to perform offset commits on a separate thread, but it is in line with what we do for sink tasks, where we leverage the Consumer::commitAsync method.

If we want to consider making offset commit synchronous (which is likely going to happen anyways when transactional writes for exactly-once source are introduced), that also might be worth a follow-up. The biggest problem I can think of with that approach would be that a single offline topic-partition would block up the entire task thread when it comes time for offset commit. If we keep the timeout for offset commit, then that'd limit the fallout and allow us to resume polling new records from the task and dispatching them to the producer after the commit attempt timed out. However, there'd still be a non-negligible throughput hit (especially for workers configured with higher offset timeouts).

@hachikuji hachikuji Feb 25, 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.

It's mostly the flushing that concerns me, not really the offset commit. I don't think we need to make it synchronous, just that it seems silly to block that shared scheduler to complete it. My thought instead was to let the scheduler trigger the flush, but then let the task be responsible for waiting for its completion. While waiting, of course, it can continue writing to outstandingMessagesBacklog. So I don't think there should be any issue from a throughput perspective.

@C0urante C0urante Mar 8, 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.

I've been ruminating over this for a few days and I think it should be possible to make task offset commits independent of each other by changing the source task offset commit scheduler to use a multi-threaded executor instead of a global single-threaded executor for all tasks. This isn't quite the same thing as what you're proposing since tasks would still not be responsible for waiting for flush completion (the offset scheduler's threads would be), but it's a smaller change and as far as I can tell, the potential downsides only really amount to a few extra threads being created.

The usage of scheduleWithFixedDelay already ensures that two offset commits for the same task won't be active at the same time, as it "Creates and executes a periodic action that becomes enabled first after the given initial delay, and subsequently with the given delay between the termination of one execution and the commencement of the next."

Beyond that, the only concern that comes to mind is potential races caused by concurrent access of the offset backing store and its underlying resources.

In distributed mode, the KafkaOffsetBackingStore and its usage of the underlying KafkaBasedLog appear to be thread-safe as everything basically boils down to calls to Producer::send, which should be fine.

In standalone mode, the MemoryOffsetBackingStore handles all writes/reads of the local offsets file via a single-threaded executor, so concurrent calls to MemoryOffsetBackingStore::set should also be fine.

Granted, none of this addresses your original concern, which is whether an offset commit timeout is necessary at all. In response to that, I think we may also want to revisit the offset commit logic and possibly do away with a timeout altogether. In sink tasks, for example, offset commit timeouts are almost a cosmetic feature at this point and are really only useful for metrics tracking. However, at the moment it's actually been pretty useful to us to monitor source task offset commit success/failure JMX metrics as a means of tracking overall task health. We might be able to make up the difference by relying on metrics for the number of active records, but it's probably not safe to make that assumption for all users, especially for what is intended to be a bug fix. So, if possible, I'd like to leave a lot of the offset commit logic intact as it is for the moment and try to keep the changes here minimal.

To summarize: I'd like to proceed by keeping the currently-proposed changes, and changing the source task offset committer to use a multi-threaded executor instead of a single-threaded executor. I can file a follow-up ticket to track improvements in offset commit logic (definitely for source tasks, and possibly for sinks) and we can look into that if it becomes a problem in the future. What do you think?

recordFlushPending = true;
offsetFlushPending = offsetWriter.beginFlush();
}
// Still wait for any producer records to flush, even if there aren't any offsets to write
// to persistent storage

Expand All @@ -490,7 +495,6 @@ public boolean commitOffsets() {
long timeoutMs = timeout - time.milliseconds();
if (timeoutMs <= 0) {
log.error("{} Failed to flush, timed out while waiting for producer to flush outstanding {} messages", this, outstandingMessages.size());
finishFailedFlush();
recordCommitFailure(time.milliseconds() - started, null);
return false;
}
Expand All @@ -506,7 +510,7 @@ public boolean commitOffsets() {
}
}

if (!flushStarted) {
if (!offsetFlushPending) {
// There was nothing in the offsets to process, but we still waited for the data in the
// buffer to flush. This is useful since this can feed into metrics to monitor, e.g.
// flush time, which can be used for monitoring even if the connector doesn't record any
Expand Down Expand Up @@ -583,15 +587,17 @@ private synchronized void finishFailedFlush() {
offsetWriter.cancelFlush();
outstandingMessages.putAll(outstandingMessagesBacklog);
outstandingMessagesBacklog.clear();
flushing = false;
recordFlushPending = false;
offsetFlushPending = false;
}

private synchronized void finishSuccessfulFlush() {
// If we were successful, we can just swap instead of replacing items back into the original map
IdentityHashMap<ProducerRecord<byte[], byte[]>, ProducerRecord<byte[], byte[]>> temp = outstandingMessages;
outstandingMessages = outstandingMessagesBacklog;
outstandingMessagesBacklog = temp;
flushing = false;
recordFlushPending = false;
offsetFlushPending = false;
}

@Override
Expand Down
135 changes: 134 additions & 1 deletion connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -841,6 +841,135 @@ public void testHeadersWithCustomConverter() throws Exception {
PowerMock.verifyAll();
}

@Test
public void testOffsetCommitIsRetriedWithSameBatchWhenRecordsRemainOutstandingWithOffsets() throws Exception {
testOffsetCommitIsRetriedWithSameBatchWhenRecordsRemainOutstanding(true);
}

@Test
public void testOffsetCommitIsRetriedWithSameBatchWhenRecordsRemainOutstandingWithoutOffsets() throws Exception {
testOffsetCommitIsRetriedWithSameBatchWhenRecordsRemainOutstanding(false);
}

private void testOffsetCommitIsRetriedWithSameBatchWhenRecordsRemainOutstanding(boolean withOffsets) throws Exception {
createWorkerTask();

EasyMock.makeThreadSafe(sourceTask, false);

sourceTask.initialize(EasyMock.anyObject(SourceTaskContext.class));
EasyMock.expectLastCall();
sourceTask.start(TASK_PROPS);
EasyMock.expectLastCall();
statusListener.onStartup(taskId);
EasyMock.expectLastCall();

sourceTask.commit();
EasyMock.expectLastCall().anyTimes();
sourceTask.commitRecord(EasyMock.anyObject(), EasyMock.anyObject());
EasyMock.expectLastCall().anyTimes();
expectConvertHeadersAndKeyValue(true);
expectApplyTransformationChain(true);
expectTaskGetTopic(true);
offsetWriter.offset(PARTITION, OFFSET);
PowerMock.expectLastCall().anyTimes();
EasyMock.expect(offsetWriter.beginFlush()).andReturn(withOffsets).anyTimes();
if (withOffsets) {
Future<Void> offsetFuture = EasyMock.niceMock(Future.class);
EasyMock.expect(offsetWriter.doFlush(EasyMock.anyObject())).andReturn(offsetFuture).anyTimes();
EasyMock.replay(offsetFuture);
}

// We'll wait for some data, then trigger a flush
EasyMock.expect(sourceTask.poll()).andAnswer(() -> {
Thread.sleep(10);
return RECORDS;
});

expectTopicCreation(TOPIC);

final CountDownLatch initialSendLatch = new CountDownLatch(1);
final Capture<org.apache.kafka.clients.producer.Callback> initialSendCallback = EasyMock.newCapture();
EasyMock.expect(producer.send(EasyMock.anyObject(), EasyMock.capture(initialSendCallback))).andAnswer(() -> {
initialSendLatch.countDown();
return null;
});

final CountDownLatch blockedPoll = new CountDownLatch(1);
EasyMock.expect(sourceTask.poll()).andAnswer(() -> {
// Block in poll here until we're ready to return the next record in the next call
blockedPoll.await();
return null;
});

// Wait for another record before flushing again
EasyMock.expect(sourceTask.poll()).andAnswer(() -> {
Thread.sleep(10);
return RECORDS;
});
final CountDownLatch subsequentSendLatch = new CountDownLatch(1);
final Capture<org.apache.kafka.clients.producer.Callback> subsequentSendCallback = EasyMock.newCapture();
EasyMock.expect(producer.send(EasyMock.anyObject(), EasyMock.capture(subsequentSendCallback))).andAnswer(() -> {
subsequentSendLatch.countDown();
return null;
});

final CountDownLatch finalPoll = new CountDownLatch(1);
EasyMock.expect(sourceTask.poll()).andAnswer(() -> {
finalPoll.await();
return null;
});

sourceTask.stop();
EasyMock.expectLastCall();

statusListener.onShutdown(taskId);
EasyMock.expectLastCall();

expectClose();

PowerMock.replayAll();


workerTask.initialize(TASK_CONFIG);
Future<?> taskFuture = executor.submit(workerTask);

assertTrue(awaitLatch(initialSendLatch));
// First offset commit fails as first record has not yet been ack'd
assertFalse(workerTask.commitOffsets());

// Unblock task and let it return the next record from poll
blockedPoll.countDown();
assertTrue(awaitLatch(subsequentSendLatch));

// Next offset commit attempt should still fail as first record has not yet been ack'd
assertFalse(workerTask.commitOffsets());

// Ack the first record
initialSendCallback.getValue().onCompletion(new RecordMetadata(new TopicPartition("foo", 0), 0, 0, 0L, 0L, 0, 0), null);

// Next offset commit attempt should succeed now as the first record has been ack'd
assertTrue(workerTask.commitOffsets());

// After a successful offset commit, the task moves onto the next batch of records, which consists solely of the second record.
// Since that record hasn't been ack'd yet, this attempt should fail
assertFalse(workerTask.commitOffsets());

// Ack the second record
subsequentSendCallback.getValue().onCompletion(new RecordMetadata(new TopicPartition("foo", 0), 0, 0, 0L, 0L, 0, 0), null);

// Next offset commit should succeed as the record has been ack'd
assertTrue(workerTask.commitOffsets());

workerTask.stop();
finalPoll.countDown();
assertTrue(workerTask.awaitStop(6000));

taskFuture.get();
assertPollMetrics(2, 2);

PowerMock.verifyAll();
}

private CountDownLatch expectEmptyPolls(int minimum, final AtomicInteger count) throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(minimum);
// Note that we stub these to allow any number of calls because the thread will continue to
Expand Down Expand Up @@ -1068,6 +1197,10 @@ private void expectOffsetFlush(boolean succeed) throws Exception {
}

private void assertPollMetrics(int minimumPollCountExpected) {
assertPollMetrics(minimumPollCountExpected, RECORDS.size());
}

private void assertPollMetrics(int minimumPollCountExpected, int activeCountMaxExpected) {
MetricGroup sourceTaskGroup = workerTask.sourceTaskMetricsGroup().metricGroup();
MetricGroup taskGroup = workerTask.taskMetricsGroup().metricGroup();
double pollRate = metrics.currentMetricValueAsDouble(sourceTaskGroup, "source-record-poll-rate");
Expand Down Expand Up @@ -1100,7 +1233,7 @@ private void assertPollMetrics(int minimumPollCountExpected) {
double activeCountMax = metrics.currentMetricValueAsDouble(sourceTaskGroup, "source-record-active-count-max");
assertEquals(0, activeCount, 0.000001d);
if (minimumPollCountExpected > 0) {
assertEquals(RECORDS.size(), activeCountMax, 0.000001d);
assertEquals(activeCountMaxExpected, activeCountMax, 0.000001d);
}
}

Expand Down