Status

Current state: Under Discussion

Discussion thread:

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

Motivation

The producer collects multiple records into batches, and a single record-specific error might fail the whole batch (e.g., `RecordTooLargeException`).

This KIP suggests adding a per-record error handler that allows users to opt into skipping bad records without failing the entire batch (similar to Kafka Streams `ProductionExceptionHandler`).

Another example for which a production exception handler could be useful is if a user tries to write into a non-existing topic, which returns a retryable error code; with infinite retries, the producer would hang retrying forever. A handler could help to break the infinite retry loop.

Public Interfaces

The interface ClientExceptionHandler and the class TransactionExceptionHandler are defined as described below.

Proposed Changes

package org.apache.kafka.common.errors;

/**
 * Interface that specifies how an exception should be handled.
 */
public interface ClientExceptionHandler {

    ClientExceptionHandlerResponse handle(final Exception exception);

    enum ClientExceptionHandlerResponse {
        /* continue processing */
        CONTINUE(0, "CONTINUE"),
        /* fail processing */
        FAIL(1, "FAIL");

        /**
         * 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;

        ClientExceptionHandlerResponse(final int id, final String name);
    }
}

package org.apache.kafka.common.errors;

/**
 * {@code ClientExceptionHandler} that continues the transaction even if a record is too large.
 * Otherwise, it makes the transaction to fail.
 */
public class TransactionExceptionHandler implements ClientExceptionHandler {

    @Override
    public ClientExceptionHandlerResponse handle(final Exception exception);
}

Compatibility, Deprecation, and Migration Plan

Changed behavior: in some cases such as having too large records, the transaction does not abort. Just all the related info is logged.

Test Plan