Skip to content

KAFKA-8586: Hot fix #239

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
Magesh Nandakumar (mageshn) merged 1 commit into from
Nov 23, 2019
Merged

KAFKA-8586: Hot fix #239

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
45 changes: 34 additions & 11 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 @@ -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.
Expand All @@ -68,6 +69,7 @@ class WorkerSourceTask extends WorkerTask {
private final OffsetStorageWriter offsetWriter;
private final Time time;
private final SourceTaskMetricsGroup sourceTaskMetricsGroup;
private final AtomicReference<Exception> producerSendException;

private List<SourceRecord> toSend;
private boolean lastSendFailed; // Whether the last send failed *synchronously*, i.e. never made it into the producer's RecordAccumulator
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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);
}
Expand All @@ -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<SourceRecord> 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
Expand All @@ -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) {
Expand Down Expand Up @@ -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;
Expand Down
51 changes: 41 additions & 10 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 @@ -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;
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -703,16 +720,24 @@ private Capture<ProducerRecord<byte[], byte[]>> expectSendRecordOnce(boolean isR
return expectSendRecordTaskCommitRecordSucceed(false, isRetry);
}

private Capture<ProducerRecord<byte[], byte[]>> expectSendRecordProducerCallbackFail() throws InterruptedException {
return expectSendRecord(false, false, false, false);
}

private Capture<ProducerRecord<byte[], byte[]>> expectSendRecordTaskCommitRecordSucceed(boolean anyTimes, boolean isRetry) throws InterruptedException {
return expectSendRecord(anyTimes, isRetry, true);
return expectSendRecord(anyTimes, isRetry, true, true);
}

private Capture<ProducerRecord<byte[], byte[]>> expectSendRecordTaskCommitRecordFail(boolean anyTimes, boolean isRetry) throws InterruptedException {
return expectSendRecord(anyTimes, isRetry, false);
return expectSendRecord(anyTimes, isRetry, true, false);
}

@SuppressWarnings("unchecked")
private Capture<ProducerRecord<byte[], byte[]>> expectSendRecord(boolean anyTimes, boolean isRetry, boolean succeed) throws InterruptedException {
private Capture<ProducerRecord<byte[], byte[]>> expectSendRecord(
boolean anyTimes,
boolean isRetry,
boolean sendSuccess,
boolean commitSuccess
) throws InterruptedException {
expectConvertKeyValue(anyTimes);
expectApplyTransformationChain(anyTimes);

Expand All @@ -729,15 +754,19 @@ private Capture<ProducerRecord<byte[], byte[]>> expectSendRecord(boolean anyTime

// 2. Converted data passed to the producer, which will need callbacks invoked for flush to work
IExpectationSetters<Future<RecordMetadata>> expect = EasyMock.expect(
producer.send(EasyMock.capture(sent),
EasyMock.capture(producerCallbacks)));
producer.send(EasyMock.capture(sent),
EasyMock.capture(producerCallbacks)));
IAnswer<Future<RecordMetadata>> expectResponse = new IAnswer<Future<RecordMetadata>>() {
@Override
public Future<RecordMetadata> 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();
}
Expand All @@ -749,8 +778,10 @@ public Future<RecordMetadata> 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;
}
Expand Down