wicknicks commented on a change in pull request #8829:
URL: https://github.com/apache/kafka/pull/8829#discussion_r438475355
##########
File path:
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExampleConnectIntegrationTest.java
##########
@@ -237,6 +239,7 @@ public void testErrantRecordReporter() throws Exception {
props.put(KEY_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
props.put(VALUE_CONVERTER_CLASS_CONFIG,
StringConverter.class.getName());
props.put(DLQ_TOPIC_NAME_CONFIG, DLQ_TOPIC);
+ props.put(ERRORS_TOLERANCE_CONFIG, ToleranceType.ALL.value());
Review comment:
minor: we should move this test to `ErrorHandlingIntegrationTest`. this
class was meant to be an example of how to do integration tests.
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
##########
@@ -556,6 +556,10 @@ private void deliverMessages() {
log.trace("{} Delivering batch of {} messages to task", this,
messageBatch.size());
long start = time.milliseconds();
task.put(new ArrayList<>(messageBatch));
+ if (retryWithToleranceOperator.failed() &&
!retryWithToleranceOperator.withinToleranceLimits()) {
Review comment:
let's add a small comment saying why we need to do this: specifically,
that if the errors raised from the operator were swallowed by the task
implementation, then here we need to kill the task, and if they were not
swallowed, we would not get here.
----------------------------------------------------------------
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
