...
We introduce the ProducerExceptionHandler interface, that which can be implemented by the user to manage the exception UnknownTopicOrPartitionException and RecordTooLargeException in the desired manner.
To configure their own handler, the user must implement the above introduced interface and add the class name in producer configuration with the key: custom.exception.handler.
It can either stop or continue the processing. Stopping processing is named as FAIL since the transaction or record sending (in non-transactional mode) will fail. In case of continuing processing either the record is dropped and the error is ignored (SWALLOW) or sending is retried (RETRY). The accepted responses for RecordTooLargeException are FAIL and SWALLOW. Therefore, RETRY will be interpreted and executed as FAIL.
To configure their own handler, the user must implement the above-introduced interface and add the class name in producer configuration with the key: custom.exception.handler.
| Code Block |
|---|
|
| Code Block |
|---|
| language | java |
|---|
| firstline | 1 |
|---|
| title | ProducerExceptionHandler |
|---|
| linenumbers | true |
|---|
|
package org.apache.kafka.common.errors;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.annotation.InterfaceStability;
/**
* Interface that specifies how an exception the RecordTooLargeException and/or UnknownTopicOrPartitionException should be handled.
*/
@InterfaceStability.Evolving
public interface ProducerExceptionHandler The accepted responses for RecordTooLargeException are FAIL and SWALLOW. Therefore, RETRY will be interpreted and executed as FAIL.
*/
@InterfaceStability.Evolving
public interface ProducerExceptionHandler extends Configurable {
/**
* Determine whether to stop processing, keep retrying internally, or swallow the error by dropping the record.
*
For RecordTooLargeException RETRY will * be interpreted and executed as FAIL.
*
* @param record The record that failed to produce
* @param exception The exception that occurred during production
*/
Response handle(final ProducerRecord<byte[], byte[]> record,
final Exception exception);
enum Response {
/* stop processing: fail */
FAIL(0, "FAIL"),
/* continue: keep retrying */
RETRY(1, "RETRY"),
/* continue: swallow the error */
SWALLOW(2, "SWALLOW");
/**
* an english description of the api--this is for debugging and can change
*/
public final String name;
/**
* the permanent and immutable id of an API--this can't change ever
*/
public final int id;
ProducerExceptionHandlerResponse(final int id,
final String name) {
this.id = id;
this.name = name;
}
}
}
|
...
| Code Block |
|---|
| language | java |
|---|
| title | ProducerConfig |
|---|
|
.
.
.
public static final String CUSTOM_EXCEPTION_HANDLER_CLASS_CONFIG = "custom.exception.handler";
private static final String CUSTOM_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.common.errors.ProducerExceptionHandler</code> interface.";.
.
.
static {
CONFIG = new ConfigDef().define(
.....
.
.
.
.define(CUSTOM_EXCEPTION_HANDLER_CLASS_CONFIG,
Type.CLASS,
null,
Importance.MEDIUM,
CUSTOM_EXCEPTION_HANDLER_CLASS_DOC);
} |
Proposed Changes
The custom handler will only affect RecordTooLargeException can be thrown by broker, producer and consumer. Of course, the ProducerExceptionHandler interface is introduced to affect ONLY the exceptions thrown from the producer. This KIP, very specifically means to provide a possibility for users to manage a limited number of exceptions (RecordTooLargeException and UnknownTopicOrPartitionException so far) thrown from the producer send() method. Of course the same exceptions may originate from different components of Apache Kafka which are not the focus of this KIP.
Notes on RecordTooLargeException based on the codebase we have today:
...
| Jira |
|---|
| server | ASF JIRA |
|---|
| serverId | 5aa69414-a9e9-3523-82ec-879b028fb15b |
|---|
| key | KAFKA-9279 |
|---|
|
...
With the changes made here (because of
| Jira |
|---|
| server | ASF JIRA |
|---|
| serverId | 5aa69414-a9e9-3523-82ec-879b028fb15b |
|---|
| key | KAFKA-9279 |
|---|
|
) the producer send() method throws a RecordTooLargeException facing too large records in transactions. The user can bring the custom handler to bear to avoid the entire batch failing by dropping the poisoning too large record (SWALLOW the error). This way, this record does not get included in the batch.
Examples
| Code Block |
|---|
|
// Example 1: RecordTooLargeException use case
// In this example we except that the producer follows the custom handler and does not fail. It may make a batch of normal records and commit the transaction successfully.
public class Example1 {
public static void main(String[] args) {
Properties producerProps = new Properties();
producerProps.put("custom.exception.handler", Example1.MyProducerExceptionHandler.class.getName());
// ..... omitted for brevity
KafkaProducer<String, String> producer = new KafkaProducer(producerProps);
Properties consumerProps = new Properties();
// ..... omitted for brevity
KafkaConsumer<String, String> consumer = new KafkaConsumer(consumerProps);
StringBuilder largeMessageStringBuffer = new StringBuilder();
for (int i = 0; i < 1000000; i++) {
largeMessageStringBuffer.append("0123456789");
}
String largeMessage = largeMessageStringBuffer.toString();
ProducerRecord<String, String> largeRecord = new ProducerRecord("output-topic", largeMessage);
ProducerRecord<String, String> normalRecord = new ProducerRecord("output-topic", "normalMessage");
producer.initTransactions();
consumer.subscribe(Collections.singleton("input-topic"));
while(true) {
try {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(60));
if (records.count() > 0) {
producer.beginTransaction();
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap();
for (ConsumerRecord<String, String> record : records) {
if (someMethod(record)) {
producer.send(largeRecord);
} else {
producer.send(normalRecord);
}
offsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1));
}
producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
producer.commitTransaction();
}
} catch (Exception e) {
producer.abortTransaction();
throw new RuntimeException(e);
}
}
}
public static class MyProducerExceptionHandler implements ProducerExceptionHandler {
@Override
public ProducerExceptionHandler.Response handle(ProducerRecord<byte[], byte[]> producerRecord, Exception e) {
if (e instanceOf RecordTooLargeException) {
return ProducerExceptionHandler.Response.SWALLOW;
}
return null;
}
@Override
public void configure(Map<String, ?> map) {
}
}
} |
...