package org.apache.kafka.clients.producer;

import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.errors.RecordTooLargeException;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;

import java.io.Closeable;

/**
 * Interface that specifies how an the RecordTooLargeException and/or UnknownTopicOrPartitionException should be handled.
 * The accepted responses for RecordTooLargeException are FAIL and SWALLOW. Therefore, RETRY will be interpreted and executed as FAIL.
 */
public interface ProducerExceptionHandler extends Configurable, Closeable {

    /**
     * Determine whether to stop processing, or swallow the error by dropping the record.
     *
     * @param record The record that failed to produce
     * @param exception The exception that occurred during production
     */
    default NonRetryableResponse handle(final ProducerRecord record, final RecordTooLargeException exception) {
		// return NonRetryableResponse.FAIL;
     the value corresponding to the default behaviour
    }

    /**
     * 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
     */
    default RetryableResponse handle(final ProducerRecord record, final UnknownTopicOrPartitionException exception) {
		// return RetryableResponse.RETRY;
	 the value corresponding to the default behaviour
    }


    enum NonRetryableResponse {
        /* stop processing: fail */
        FAIL(0, "FAIL"),
        /* drop the record and continue */
        SWALLOW(1, "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;

        RecordTooLargeExceptionResponse(final int id, final String name) {
            this.id = id;
            this.name = name;
        }
    }
    enum RetryableResponse {
        /* 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;

        UnknownTopicOrPartitionExceptionResponse(final int id, final String name) {
            this.id = id;
            this.name = name;
        }
    }
}  

.
.
.

public static final String CUSTOM_EXCEPTION_HANDLER_CLASS_CONFIG = "custom.exception.handler.class";
private static final String CUSTOM_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.common.errors.ProducerExceptionHandler</code> interface.";.

public static final String DROP_INVALID_LARGE_RECORDS_CONFIG = "drop.invalid.large.records";
private static final String DROP_INVALID_LARGE_RECORDS_DOC = "When set to 'true', records larger than <code>" + MAX_REQUEST_SIZE_CONFIG + "</code> are dropped."

public static final String RETRY_UNKNOWN_TOPIC_PARTITION_MS_CONFIG = "retry.unknown.topic.partition.ms";
private static final String RETRY_UNKNOWN_TOPIC_PARTITION_MS_DOC = "When set to 'false', retry is not done by producer when encountering UnknownTopicOrPartiitonException. Otherwise, it retries for The configuration controls the retrying duration encountering the UnknownTopicOrPartitionException. The user should set it to lower the timeout compared to <code>" + MAX_BLOCK_MS_CONFIG + "</code> ms.";.
. .
.
static {
CONFIG = new ConfigDef().define(
.....
.
.
.
                         .define(CUSTOM_EXCEPTION_HANDLER_CLASS_CONFIG,
                                 Type.CLASS,
                                 null,
                                 Importance.MEDIUM,
                                 CUSTOM_EXCEPTION_HANDLER_CLASS_DOC)
                         .define(DROP_INVALID_LARGE_RECORDS_CONFIG,
                                 Type.BOOLEAN,
                                 false,
                                 Importance.LOW,
                                 DROP_INVALID_LARGE_RECORDS_DOC)
                        .define(RETRY_UNKNOWN_TOPIC_PARTITION_MS_CONFIG,
                                 Type.BOOLEANINT,
                                 true Integer.MAX_VALUE,
                                 Importance.LOW,
                                 RETRY_UNKNOWN_TOPIC_PARTITION_MS_DOC);
 }