Status

Current state: Under Discussion

Discussion thread: here

JIRA: KAFKA-15309 
Other related tickets: KAFKA-9279, KAFKA-15259

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

We believe that the user should be able to develop custom exception handlers for managing producer exceptions. On the other hand, this will be an expert-level API, and using that may result in strange behaviour in the system, making it hard to find the root cause. Therefore, the custom handler is currently limited to handling RecordTooLargeException and UnknownTopicOrPartitionException. The motivation for this KIP is derived from the following use cases:

• In transactions, the producer collects multiple records in batches. Then a RecordTooLargeException related to a single record leads to failing the entire batch. A custom exception handler in this case may decide on dropping the record and continuing the processing.
• When a user tries to write into a non-existing topic, it returns a retryable error code; with infinite retries, the producer would hang retrying forever. A custom handler in this case could help to break the infinite retry loop.

This KIP introduces an interface that can be implemented by the user to handle the exceptions UnknownTopicOrPartitionException and RecordTooLargeException.

Question: Why do we need an interface for handling the exceptions? Could we have a couple of simple producer configuration options for those two exceptions? 
Answer: We aim at giving the user the flexibility of an interface. For example, facing UnknownTopicOrPartitionException, the user may want to raise an error for some topics but retry it for other topics. Having a configuration option with a fixed set of possibilities does not serve the user's needs.   

Public Interfaces

We introduce the ProducerExceptionHandler interface, that can be implemented by the user to manage the exception 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.

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 should be handled.
 */
@InterfaceStability.Evolving
public interface ProducerExceptionHandler extends Configurable {

    /**
     * Determine whether to stop processing, keep retrying internally, or swallow the error by dropping the record.
     *
     * @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;
        }
    }
}
 

.
.
.

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 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:

• When producer sends a too large record in non-transactional mode, the producer send() method throws no exception but returns a record metadata that includes the error RecordTooLargeException. Obviously, a failed sent record does not reach the broker.  
• With the changes made here (because of KAFKA-9279 - Getting issue details... STATUS ) 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.
• There is another scenario in which the record size is acceptable by the producer (the record is NOT too large from producer point of view due to setting the "max.request.size" and "buffer.memory" to big numbers in producer config), but it is too large for the broker (The default message size of broker is 1 MB). In such case, the broker throws RecordTooLargeException during commitTransaction(). This scenario is not the focus of this KIP.

Examples

// Example 1: RecordTooLargeException use case 
public class Example1 {  
	public static void main(String[] args) {
        
		Properties producerProps = new Properties();
        ..... // omitted for brevity
	    producerProps.put("custom.exception.handler", Example1.ContinueTransaction.class.getName());
        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 largeMessge = largeMessageStringBuffer.toString();

        
		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) {
                        ProducerRecord<String, String> largeRecord = new ProducerRecord("output-topic", largeMessge);
						Future<RecordMetadata> send = producer.send(largeRecord);
                        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 ContinueTransaction implements ProducerExceptionHandler {
        @Override
        public ProducerExceptionHandler.Response handle(ProducerRecord<byte[], byte[]> producerRecord, Exception e) {
            return ProducerExceptionHandler.Response.SWALLOW;
        }
        @Override
        public void configure(Map<String, ?> map) {

        }
    }
}

Compatibility, Deprecation, and Migration Plan

Changed behaviour: The default behaviour stays as it is, but the user can change the behaviour by implementing the handle() function.

Test Plan

Some unit and integration tests will be implemented to ensure that

• exceptions are caught by the ProducerExceptionHandler and the provided implementations.
• the right exceptions are caught.

No unit tests are needed to ensure the backward compatibility. Passing the current unit tests is an enough indicator.

Rejected Alternatives

Using one or more producer configs instead of having a pluggable interface: misusing produce configs has the same drawbacks of misusing the interface while the interface solution provides a handler with the advantage of full flexibility. Later, further KIPs can be proposed to cover more exceptions or more actions for handing.