THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
This feature will directly make changes to the SinkTaskContext class with the addition of a new method and a new interface.
Method
This KIP will add a getter method to the SinkTaskContext class that will return a error reporter object. Sink connectors that wish to use a error reporter can call this method within their tasks.
...
| Code Block |
|---|
private BiFunction<SinkTask, Throwable, Future<Void> reporter;
@Override
public void start(Map<String, String> props) {
...
try {
reporter = context.failedRecordReporter(); // may be null if DLQ not enabled
} catch (NoSuchMethodException e) {
// Will occur in Connect runtimes earlier than 2.9
reporter = null;
}
}
@Override
public void put(Collection<SinkRecord> records) {
for (SinkRecord record: records) {
try {
// attempt to send record to data sink
process(record);
} catch(Exception e) {
if (reporter != null) {
// Send errant record to error reporter
Future<Void> future = reporter.accept(record, e);
// Optionally wait till the failure's been recorded in Kafka
future.get();
} else {
// There's no error reporter, so fail
throw new ConnectException("Failed on record", e);
}
}
}
} |
Interface
The interface will look like the following:
| Code Block |
|---|
/** * Component that the sink task can use as it {@link SinkTask#put(/** * Reporter of problematic records and the corresponding problems. * * @since 2.9 */ public interface ErrantRecordReporter { /** * Report a problematic record and the corresponding error to be written to the sink * connector's dead letter queue (DLQ). * * <p>This call is asynchronous and returns a {@link java.util.concurrent.Future Future} for the * {@link RecordMetadata} that will be assigned to the record in the DLQ topic. Invoking * {@link java.util.concurrent.Future#get() get()} on this future will block until the * record has been written and then return the metadata for the record * or throw any exception that occurred while sending the record. * If you want to simulate a simple blocking call you can call the <code>get()</code> method * immediately. * * @param record the problematic record; may not be null * @param error the error capturing the problem with the record; may not be null * @return a future that can be used to block until the record and error are written * to the DLQ * @since 2.9 */ default Future<RecordMetadata> report(SinkRecord record, Throwable error) { return report(record, error, null); } /** * Report a problematic record and the corresponding error to be written to the sink * connector's dead letter queue (DLQ). * * <p>This call is asynchronous and returns a {@link java.util.concurrent.Future Future} for the * {@link RecordMetadata} that will be assigned to the record in the DLQ topic. Invoking * {@link java.util.concurrent.Future#get() get()} on this future will block until the * record has been written and then return the metadata for the record * or throw any exception that occurred while sending the record. * If you want to simulate a simple blocking call you can call the <code>get()</code> method * immediately. * * <p>Fully non-blocking usage can make use of the {@link Callback} |
Proposed Changes
...
parameter to provide a
* callback that will be invoked when the request is complete. Callbacks for records being
* sent to the same partition are guaranteed to execute in order.
*
* @param record the problematic record; may not be null
* @param error the error capturing the problem with the record; may not be null
* @param callback A user-supplied callback to execute when the record has been acknowledged
* by the server; may be null for no callback
* @return a future that can be used to block until the record and error are written
* to the DLQ
* @since 2.9
*/
Future<RecordMetadata> report(SinkRecord record, Throwable error, Callback callback);
} |
Proposed Changes
Reporting
The SinkTaskContext class will create a BiFunction<SinkRecord, Throwable, Future<Void>> object as the error reporter object. Along with the original errant sink record, the exception thrown due to the errant sink record will be passed to the error reporter to provide additional context. For asynchronous functionality, the error reporter will return a Future for each record processed. The error reporter will use the same configurations as the dead letter queue in KIP-298 to avoid redundant configuration. There will be no additional configurations for the Producer and AdminClient under the hood aside from the existing producer. and admin. configurations present in the worker configurations and producer.override. and admin.override. configurations present in the connector configurations. Serialization for the errant records will be done in the same manner as KIP-298.
...