Skip to content

KAFKA-13348: Allow Source Tasks to Handle Producer Exceptions #11382

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
mimaison merged 5 commits into from
Jan 27, 2022
Merged

KAFKA-13348: Allow Source Tasks to Handle Producer Exceptions #11382

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
34d93a8
KAFKA-13348 errors.tolerance "all" behavior has been updated to allow…
TheKnowles Nov 19, 2021
b956c19
KAFKA-13348 Fixed related failing test after lastSendFail was removed.
TheKnowles Nov 19, 2021
22daf36
KAFKA-13348 Updated error tolerance in WorkerSourceTask to use existi…
TheKnowles Dec 6, 2021
878b131
KAFKA-13348 Added additional connect worker logging. Cleaned up worke…
TheKnowles Jan 26, 2022
ad098eb
KAFKA-13348 Updated logging for error ignore path.
TheKnowles Jan 27, 2022
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
9 changes: 5 additions & 4 deletions connect/api/src/main/java/org/apache/kafka/connect/source/SourceTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,8 +115,9 @@ public void commitRecord(SourceRecord record) throws InterruptedException {
/**
* <p>
* Commit an individual {@link SourceRecord} when the callback from the producer client is received. This method is
* also called when a record is filtered by a transformation, and thus will never be ACK'd by a broker. In this case
* {@code metadata} will be null.
* also called when a record is filtered by a transformation or when {@link ConnectorConfig} "errors.tolerance" is set to "all"
* and thus will never be ACK'd by a broker.
* In both cases {@code metadata} will be null.
* </p>
* <p>
* SourceTasks are not required to implement this functionality; Kafka Connect will record offsets
Expand All @@ -128,8 +129,8 @@ public void commitRecord(SourceRecord record) throws InterruptedException {
* not necessary to implement both methods.
* </p>
*
* @param record {@link SourceRecord} that was successfully sent via the producer or filtered by a transformation
* @param metadata {@link RecordMetadata} record metadata returned from the broker, or null if the record was filtered
* @param record {@link SourceRecord} that was successfully sent via the producer, filtered by a transformation, or dropped on producer exception
* @param metadata {@link RecordMetadata} record metadata returned from the broker, or null if the record was filtered or if producer exceptions are ignored
* @throws InterruptedException
*/
public void commitRecord(SourceRecord record, RecordMetadata metadata)
Expand Down
7 changes: 6 additions & 1 deletion 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 @@ -39,6 +39,7 @@
import org.apache.kafka.connect.runtime.distributed.ClusterConfigState;
import org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator;
import org.apache.kafka.connect.runtime.errors.Stage;
import org.apache.kafka.connect.runtime.errors.ToleranceType;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTask;
import org.apache.kafka.connect.storage.CloseableOffsetStorageReader;
Expand Down Expand Up @@ -366,7 +367,11 @@ private boolean sendRecords() {
if (e != null) {
log.error("{} failed to send record to {}: ", WorkerSourceTask.this, topic, e);

@C0urante C0urante Nov 19, 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.

Should we modify this line to respect the errors.log.enable (and possibly errors.log.include.messages) properties?

I wonder if it might be useful to still unconditionally set producerSendException (or perhaps even convert that field from an AtomicReference<Throwable> to some kind of list, and append to it here) and then modify the contents (and possibly also name) of maybeThrowProducerSendException to have our error-handling logic. Thoughts?

@C0urante C0urante Nov 19, 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.

Actually, that may complicate things by causing records to be given to SourceTask::commitRecord out of order (a record that caused a producer failure may be committed after a record that was dispatched to the producer after it). So probably best to keep the error-handling logic here, but I do still wonder if we can respect the logging-related configuration properties.

@TheKnowles TheKnowles Nov 19, 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.

Now that this could be a tolerated error, it makes sense to have it respect the errors.log.enable configuration, but the log line would be duplicated, unconditionally writing it in the event we do not tolerate and a config check if we do.

Are you envisioning something like this?

if (retryWithToleranceOperator.getErrorToleranceType().equals(ToleranceType.ALL)) {
    if (errorLogEnabled) { // get this value from the config in some manner
        log.error("{} failed to send record to {}: ", WorkerSourceTask.this, topic, e);
        log.trace("{} Failed record: {}", WorkerSourceTask.this, preTransformRecord);
    }
    commitTaskRecord(preTransformRecord, null);
} else {
    log.error("{} failed to send record to {}: ", WorkerSourceTask.this, topic, e);
    log.trace("{} Failed record: {}", WorkerSourceTask.this, preTransformRecord);
    producerSendException.compareAndSet(null, e);
}

I would need to look more closely at the other layers of objects on top of the SourceTask. enableErrorLog() is available in the ConnectorConfig, but only the SinkConnectorConfig makes use of it. I would need to spin up some additional infrastructure. Not sure if I would want to add WorkerErrantRecordReporter to WorkerSourceTask or have the configuration pass down in some other manner.

@C0urante C0urante Nov 22, 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.

Yes, I was thinking the behavior could be something like that code snippet, although we'd also want to respect the errors.log.include.messages property and would probably want the format of the error messages to be similar to the error messages we emit in other places where messages are tolerated (such as when conversion or transformation fails).

@TheKnowles TheKnowles Dec 6, 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.

The error retry handling infrastructure predominantly concerns itself with the sink side of the house. Insofar that any refactoring I would want to do would probably necessitate a KIP on its own. To that end, I have added an additional executeFailed() function to RetryWithToleranceOperator to allow the source worker to handle error logging with all of the existing infrastructure/configuration that exists for sink tasks.

I toy'ed around with the idea of having the new executeFailed() fire without a tolerance type check. This would work for failing/ignoring as expected, but with no mechanism to then decide if we should call commitRecord(). We could block on the future from executeFailed() and then check withinToleranceLimits() but that introduces non determinism with interrupt/execution exceptions.

log.trace("{} Failed record: {}", WorkerSourceTask.this, preTransformRecord);
producerSendException.compareAndSet(null, e);
if (retryWithToleranceOperator.getErrorToleranceType().equals(ToleranceType.ALL)) {

@mimaison mimaison Jan 26, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

We can use == to compare enums.

@TheKnowles TheKnowles Jan 26, 2022

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

commitTaskRecord(preTransformRecord, null);

@mimaison mimaison Jan 26, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Should we have a debug/trace log in this path?

@TheKnowles TheKnowles Jan 26, 2022

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.

Previously it was suggested to have the tolerance operator handle via the logging report. I would personally find it useful to have it in the connect log regardless of tolerance error logging configuration. I've moved the error/debug log lines to above the tolerance check to log in all instances.

@C0urante C0urante Jan 26, 2022

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.

We should not be logging at ERROR level for every single record if we aren't failing the task unless the user has explicitly enabled this by setting errors.log.enable to true in their connector config.

@mimaison mimaison Jan 27, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Let's keep the existing trace and error log lines in the else block.
My suggestion is to add a line at the debug or trace level in the if block so users can know if an error is ignored.

@TheKnowles TheKnowles Jan 27, 2022

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.

My misunderstanding, thank you both for the feedback. Update made.

} else {
producerSendException.compareAndSet(null, e);
}
} else {
submittedRecord.ack();
counter.completeRecord();
Expand Down
7 changes: 7 additions & 0 deletions ...ime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -229,6 +229,13 @@ public synchronized boolean withinToleranceLimits() {
}
}

// For source connectors that want to skip kafka producer errors.
// They cannot use withinToleranceLimits() as no failure may have actually occurred prior to the producer failing
// to write to kafka.
public synchronized ToleranceType getErrorToleranceType() {

@mimaison mimaison Jan 26, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Does this need to be synchronized?

@TheKnowles TheKnowles Jan 26, 2022

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 does not. Type is immutable and thread safe. I had dug through the ticket that retroactively made this class thread safe and it seemed like a good idea at the time to slap a synchronized on it to match the rest of the class, but is not necessary at all. Removed.

return errorToleranceType;
}

// Visible for testing
boolean checkRetry(long startTime) {
return (time.milliseconds() - startTime) < errorRetryTimeout;
Expand Down
34 changes: 34 additions & 0 deletions 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 @@ -222,6 +222,13 @@ private void createWorkerTask() {
createWorkerTask(TargetState.STARTED);
}

private void createWorkerTaskWithErrorToleration() {

@mimaison mimaison Jan 26, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Can we reuse the createWorkerTask() method just below by passing a RetryWithToleranceOperator argument instead of creating the WorkerSourceTask object here?

@TheKnowles TheKnowles Jan 26, 2022

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 have refactored the constructors to be cleaner with various parameter lists.

workerTask = new WorkerSourceTask(taskId, sourceTask, statusListener, TargetState.STARTED, keyConverter, valueConverter, headerConverter,
transformationChain, producer, admin, TopicCreationGroup.configuredGroups(sourceConfig),
offsetReader, offsetWriter, config, clusterConfigState, metrics, plugins.delegatingLoader(), Time.SYSTEM,
RetryWithToleranceOperatorTest.ALL_OPERATOR, statusBackingStore, Runnable::run);
}

private void createWorkerTask(TargetState initialState) {
createWorkerTask(initialState, keyConverter, valueConverter, headerConverter);
}
Expand Down Expand Up @@ -815,6 +822,33 @@ public void testSendRecordsTaskCommitRecordFail() throws Exception {
PowerMock.verifyAll();
}

@Test
public void testSourceTaskIgnoresProducerException() throws Exception {
createWorkerTaskWithErrorToleration();
expectTopicCreation(TOPIC);

// send two records
// record 1 will succeed
// record 2 will invoke the producer's failure callback, but ignore the exception via retryOperator
// and no ConnectException will be thrown
SourceRecord record1 = new SourceRecord(PARTITION, OFFSET, TOPIC, 1, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD);
SourceRecord record2 = new SourceRecord(PARTITION, OFFSET, TOPIC, 2, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD);


expectSendRecordOnce();
expectSendRecordProducerCallbackFail();
sourceTask.commitRecord(EasyMock.anyObject(SourceRecord.class), EasyMock.anyObject(RecordMetadata.class));

@mimaison mimaison Jan 26, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Instead of EasyMock.anyObject(RecordMetadata.class) should we use EasyMock.isNull() to assert we indeed pass null to the task in case there was a failure?

@TheKnowles TheKnowles Jan 26, 2022

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

EasyMock.expectLastCall();

PowerMock.replayAll();

Whitebox.setInternalState(workerTask, "toSend", Arrays.asList(record1, record2));
Whitebox.invokeMethod(workerTask, "sendRecords");
assertEquals(false, Whitebox.getInternalState(workerTask, "lastSendFailed"));

PowerMock.verifyAll();
}

@Test
public void testSlowTaskStart() throws Exception {
final CountDownLatch startupLatch = new CountDownLatch(1);
Expand Down
7 changes: 7 additions & 0 deletions ...src/test/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,8 @@ public class RetryWithToleranceOperatorTest {

public static final RetryWithToleranceOperator NOOP_OPERATOR = new RetryWithToleranceOperator(
ERRORS_RETRY_TIMEOUT_DEFAULT, ERRORS_RETRY_MAX_DELAY_DEFAULT, NONE, SYSTEM);
public static final RetryWithToleranceOperator ALL_OPERATOR = new RetryWithToleranceOperator(
ERRORS_RETRY_TIMEOUT_DEFAULT, ERRORS_RETRY_MAX_DELAY_DEFAULT, ALL, SYSTEM);
static {
Map<String, String> properties = new HashMap<>();
properties.put(CommonClientConfigs.METRICS_NUM_SAMPLES_CONFIG, Objects.toString(2));
Expand All @@ -92,6 +94,11 @@ public class RetryWithToleranceOperatorTest {
new ConnectorTaskId("noop-connector", -1),
new ConnectMetrics("noop-worker", new TestableWorkerConfig(properties), new SystemTime(), "test-cluster"))
);
ALL_OPERATOR.metrics(new ErrorHandlingMetrics(
new ConnectorTaskId("errors-all-tolerate-connector", -1),
new ConnectMetrics("errors-all-tolerate-worker", new TestableWorkerConfig(properties),
new SystemTime(), "test-cluster"))
);
}

@SuppressWarnings("unused")
Expand Down