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.

Public Interfaces

We propose the following new public interface:

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

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

/**
 * Provides a mechanism for reporting errors
 * using the information contained in an `ErrorContext`.
 *
 * @param <T> the type of the 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() {
	}
}

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

class name: `org.apache.kafka.connect.reporter.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

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.reporters.$alias.type` – fully qualified class name for the error reporter  
- `errors.reporters.$alias.*` – all other keys as defined in `ErrorRecordReporter.config()` are prefixed with this alias.

Example:

errors.reporters=example

errors.reporters.example.type=com.example.SimpleErrorRecordReporter

errors.reporters.example.param=testValue

Proposed Changes

We 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/apache/kafka/pull/17493

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.