Status

Current state: Under Discussion

Discussion thread: here

JIRA: KAFKA-15309 

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

Modyfying affected unit tests as well as adding an integration test for all the affected cases.

Rejected Alternatives