From b4b53e9bd28ff6a956f49b3f5e87581ba8e49b19 Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Sun, 25 Aug 2019 13:54:00 -0700 Subject: [PATCH] KAFKA-8586: Fail source tasks when producers fail to send records (#6993) Changed Connect's `WorkerSourceTask` to capture non-retriable exceptions from the `producer.send(...)` (e.g., authentication or authorization errors) and to fail the connector task when such an error is encountered. Modified the existing unit tests to verify this functionality. Note that most producer errors are retriable, and Connect will (by default) set up each producer with 1 max in-flight message and infinite retries. This change only affects non-retriable errors. (cherry picked from commit 3aa9f99) --- .../connect/runtime/WorkerSourceTask.java | 45 ++++++++++++---- .../connect/runtime/WorkerSourceTaskTest.java | 51 +++++++++++++++---- 2 files changed, 75 insertions(+), 21 deletions(-) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java index 27dd388916879..7ca564cf9c21f 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java @@ -49,6 +49,7 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicReference; /** * WorkerTask that uses a SourceTask to ingest data into Kafka. @@ -68,6 +69,7 @@ class WorkerSourceTask extends WorkerTask { private final OffsetStorageWriter offsetWriter; private final Time time; private final SourceTaskMetricsGroup sourceTaskMetricsGroup; + private final AtomicReference producerSendException; private List toSend; private boolean lastSendFailed; // Whether the last send failed *synchronously*, i.e. never made it into the producer's RecordAccumulator @@ -117,6 +119,7 @@ public WorkerSourceTask(ConnectorTaskId id, this.flushing = false; this.stopRequestedLatch = new CountDownLatch(1); this.sourceTaskMetricsGroup = new SourceTaskMetricsGroup(id, connectMetrics); + this.producerSendException = new AtomicReference<>(); } @Override @@ -198,10 +201,12 @@ public void execute() { continue; } + maybeThrowProducerSendException(); + if (toSend == null) { log.trace("{} Nothing to send to Kafka. Polling source for additional records", this); long start = time.milliseconds(); - toSend = task.poll(); + toSend = poll(); if (toSend != null) { recordPollReturned(toSend.size(), time.milliseconds() - start); } @@ -223,7 +228,27 @@ public void execute() { } } - /** + private void maybeThrowProducerSendException() { + if (producerSendException.get() != null) { + throw new ConnectException( + "Unrecoverable exception from producer send callback", + producerSendException.get() + ); + } + } + + protected List poll() throws InterruptedException { + try { + return task.poll(); + } catch (RetriableException | org.apache.kafka.common.errors.RetriableException e) { + log.warn("{} failed to poll records from SourceTask. Will retry operation.", this, e); + // Do nothing. Let the framework poll whenever it's ready. + return null; + } + } + + + /** * Try to send a batch of records. If a send fails and is retriable, this saves the remainder of the batch so it can * be retried after backing off. If a send fails and is not retriable, this will throw a ConnectException. * @return true if all messages were sent, false if some need to be retried @@ -233,6 +258,8 @@ private boolean sendRecords() { recordBatch(toSend.size()); final SourceRecordWriteCounter counter = new SourceRecordWriteCounter(toSend.size(), sourceTaskMetricsGroup); for (final SourceRecord preTransformRecord : toSend) { + maybeThrowProducerSendException(); + final SourceRecord record = transformationChain.apply(preTransformRecord); if (record == null) { @@ -269,26 +296,22 @@ private boolean sendRecords() { @Override public void onCompletion(RecordMetadata recordMetadata, Exception e) { if (e != null) { - // Given the default settings for zero data loss, this should basically never happen -- - // between "infinite" retries, indefinite blocking on full buffers, and "infinite" request - // timeouts, callbacks with exceptions should never be invoked in practice. If the - // user overrode these settings, the best we can do is notify them of the failure via - // logging. - log.error("{} failed to send record to {}: {}", WorkerSourceTask.this, topic, e); + log.error("{} failed to send record to {}:", WorkerSourceTask.this, topic, e); log.debug("{} Failed record: {}", WorkerSourceTask.this, preTransformRecord); + producerSendException.compareAndSet(null, e); } else { + recordSent(producerRecord); + counter.completeRecord(); log.trace("{} Wrote record successfully: topic {} partition {} offset {}", WorkerSourceTask.this, recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset()); commitTaskRecord(preTransformRecord); } - recordSent(producerRecord); - counter.completeRecord(); } }); lastSendFailed = false; - } catch (RetriableException e) { + } catch (org.apache.kafka.common.errors.RetriableException e) { log.warn("{} Failed to send {}, backing off before retrying:", this, producerRecord, e); toSend = toSend.subList(processed, toSend.size()); lastSendFailed = true; diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java index dfdf415e0ed3a..193f8772f5f8d 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java @@ -21,9 +21,11 @@ import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.MetricName; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.TopicAuthorizationException; import org.apache.kafka.common.record.InvalidRecordException; import org.apache.kafka.common.utils.Time; import org.apache.kafka.connect.data.Schema; +import org.apache.kafka.connect.errors.ConnectException; import org.apache.kafka.connect.runtime.ConnectMetrics.MetricGroup; import org.apache.kafka.connect.runtime.WorkerSourceTask.SourceTaskMetricsGroup; import org.apache.kafka.connect.runtime.isolation.Plugins; @@ -534,6 +536,21 @@ public void testSendRecordsRetries() throws Exception { PowerMock.verifyAll(); } + @Test(expected = ConnectException.class) + public void testSendRecordsProducerCallbackFail() throws Exception { + createWorkerTask(); + + 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); + + expectSendRecordProducerCallbackFail(); + + PowerMock.replayAll(); + + Whitebox.setInternalState(workerTask, "toSend", Arrays.asList(record1, record2)); + Whitebox.invokeMethod(workerTask, "sendRecords"); + } + @Test public void testSendRecordsTaskCommitRecordFail() throws Exception { createWorkerTask(); @@ -703,16 +720,24 @@ private Capture> expectSendRecordOnce(boolean isR return expectSendRecordTaskCommitRecordSucceed(false, isRetry); } + private Capture> expectSendRecordProducerCallbackFail() throws InterruptedException { + return expectSendRecord(false, false, false, false); + } + private Capture> expectSendRecordTaskCommitRecordSucceed(boolean anyTimes, boolean isRetry) throws InterruptedException { - return expectSendRecord(anyTimes, isRetry, true); + return expectSendRecord(anyTimes, isRetry, true, true); } private Capture> expectSendRecordTaskCommitRecordFail(boolean anyTimes, boolean isRetry) throws InterruptedException { - return expectSendRecord(anyTimes, isRetry, false); + return expectSendRecord(anyTimes, isRetry, true, false); } - @SuppressWarnings("unchecked") - private Capture> expectSendRecord(boolean anyTimes, boolean isRetry, boolean succeed) throws InterruptedException { + private Capture> expectSendRecord( + boolean anyTimes, + boolean isRetry, + boolean sendSuccess, + boolean commitSuccess + ) throws InterruptedException { expectConvertKeyValue(anyTimes); expectApplyTransformationChain(anyTimes); @@ -729,15 +754,19 @@ private Capture> expectSendRecord(boolean anyTime // 2. Converted data passed to the producer, which will need callbacks invoked for flush to work IExpectationSetters> expect = EasyMock.expect( - producer.send(EasyMock.capture(sent), - EasyMock.capture(producerCallbacks))); + producer.send(EasyMock.capture(sent), + EasyMock.capture(producerCallbacks))); IAnswer> expectResponse = new IAnswer>() { @Override public Future answer() throws Throwable { synchronized (producerCallbacks) { for (org.apache.kafka.clients.producer.Callback cb : producerCallbacks.getValues()) { - cb.onCompletion(new RecordMetadata(new TopicPartition("foo", 0), 0, 0, - 0L, 0L, 0, 0), null); + if (sendSuccess) { + cb.onCompletion(new RecordMetadata(new TopicPartition("foo", 0), 0, 0, + 0L, 0L, 0, 0), null); + } else { + cb.onCompletion(null, new TopicAuthorizationException("foo")); + } } producerCallbacks.reset(); } @@ -749,8 +778,10 @@ public Future answer() throws Throwable { else expect.andAnswer(expectResponse); - // 3. As a result of a successful producer send callback, we'll notify the source task of the record commit - expectTaskCommitRecord(anyTimes, succeed); + if (sendSuccess) { + // 3. As a result of a successful producer send callback, we'll notify the source task of the record commit + expectTaskCommitRecord(anyTimes, commitSuccess); + } return sent; }