Status

Current state: Under Discussion

Discussion thread: here

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

Motivation

...

Kafka Connect currently supports

...

basic error tolerance mechanisms (`none`, `all`) and dead letter queues (DLQ) for error reporting. However, there is no flexible way for developers to define how errors should be handled during specific stages of record processing.

Kafka Streams already provides this flexibility via the ProcessingExceptionHandler interface, which allows stream applications to determine whether to continue, fail, or log on processing errors. This KIP proposes a similar mechanism for Connect: a pluggable `ErrorHandler` interface for custom error handling logic.

Proposed changes

A new `ErrorHandler` interface will be added to the Kafka Connect API. It will allow the Connect framework to delegate error handling to a plugin-defined handler. 

This interface is inspired by the existing `ProcessingExceptionHandler` interface in Kafka Streams, which gives users control over how to react to processing errors in stream tasks. Kafka Connect’s `ErrorHandler` brings a similar level of flexibility to Connect pipelines.

- `org.apache.kafka.connect.runtime.errors.DeadLetterQueueReporter` - Reports records into a dead letter queue.

- `org.apache.kafka.connect.runtime.errors.LogReporter` - An abstract class for reporting errors via logging. Both sink and source connectors have their own implementations of the log reporter.

In addition to these implementations, custom error reporters can be beneficial for implementing custom logic to report errors to specific storage systems, as well as defining custom metrics and logs based on the error context. This can be particularly useful for source connectors, where the log reporter is the only option to understand the root cause of failures. This proposal aims to introduce the capability to define custom error reporting functionality.

Public Interfaces

...

package: `org.apache.kafka.connect.

...

handler;`

class name: `org.apache.kafka.connect.reporter.

...

ErrorHandler`

public interface ErrorHandler<T> {

    /**
  *    Provides  a* mechanismHandle forthe reportingexception errors
that *occurred usingduring thea informationspecific containedstage inof anrecord `ErrorContext`processing.
        *
  *  @param  <T>  the* type@param ofcontext theprocessing error context.
     */
public interface ErrorRecordReporter<T> extends Configurable, AutoCloseable {

	/**
	 * Report an error using the provided error context.
	 *
	 * @param context the error context (cannot be null)
	 */
	void report(ErrorContext<T> context);

	@Override
	default void close() {
	}
}    ErrorHandlerResponse handleError(ErrorContext<T> context);

    enum ErrorHandlerResponse {
        DROP,     // Silently skip the record.
        FAIL,       // Fail the task.
        ACK        // Acknowledge and skip; relevant for source connectors.
     }
}

package: `org.apache.kafka.connect.reporterhandler;`

class name: `org.apache.kafka.connect.reporterhandler.ErrorContext`

public class ErrorContext<T> {

	private final String stage;
	private final String executingClassName;
	private final T original;
	private final Throwable error;

	public ErrorContext(String stage, String executingClassName, T original, Throwable error) {
		this.stage = stage;
		this.executingClassName = executingClassName;
		this.original = original;
		this.error = error;
	}

	public String stage() {
		return stage;
	}

	public String executingClassName() {
		return executingClassName;
	}

	public T original() {
		return original;
	}

	public Throwable error() {
		return error;
	}

}

Additionally, we propose to add the following configuration properties.

Connector Configuration

Extend the existing errors.tolerance config to support a new mode:

• `none` – current behavior.

• `all` – current behavior.

• `custom` – use a pluggable `ErrorHandler`.

Introduces a configuration key to specify the fully qualified class name of the `ErrorHandler`. 

- `errors.handler` – name of error handler.The error record reporter chain will be configured at the connector-level. The order of the error record reporters is defined by the `errors.reporters` config which represents a list of aliases. Each alias in `errors.reporters` implies that some additional keys are configurable:  
- `errors.reportershandler.$alias.type` – fully qualified class name for the error reporter  handler.  
- `errors.reportershandler.$alias.*` – all other keys as defined in `ErrorRecordReporterErrorHandler.config()` are prefixed with this alias.

...

errors.reportershandler=example

errors.reportershandler.example.type=com.example.SimpleErrorRecordReporterSimpleErrorHandler

errors.reportershandler.example.param=testValue

Proposed Changes

Integration with RetryWithToleranceOperator

When `errors.tolerance=custom`, the Connect runtime will:

• Instantiate and configure the specified `ErrorHandler`.

• Delegate error handling for supported stages to the plugin.

• Act based on the returned `ErrorHandlerResponse`.

For example:

• If `DROP`: record is silently skipped.

• If `FAIL`: the task is failed.

• If `ACK`: record is acknowledged as consumed (only for source connectors).

Small POCWe propose a new pluggable interface (`ErrorRecordReporter`) which enables the creation of custom error record reporters. These custom reporters will be executed when `errors.tolerance=all`. The draft pull request: https://github.com/apacheanton-liauchuk/kafka/pull/174931/files

Compatibility, Deprecation, and Migration Plan

There are no backward compatibility concerns.

Rejected Alternatives

1. Exposing `ErrorReporter` and `ProcessingContext` as public APIs: The main issue with this design is that it exposes packages with `runtime` in their names, which cannot be changed.
2. Naming as ErrorReporter: This name is already used for error reporters in the runtime module. To avoid name conflicts, `ErrorRecordReporter` was chosen Extending the Reporter interface: Considered, but it conflates reporting and control logic. Keeping handling and reporting responsibilities separate promotes clarity and composability.