自定义Kafka Sink连接器错误处理配置失效，需实现特定接口吗？

是的，你需要调整自定义连接器的异常处理逻辑，或实现特定接口，才能让Kafka Connect的errors.tolerance=all配置生效。

核心原因 #

Kafka Connect的错误容忍机制仅对单条记录的处理异常（比如数据转换失败、写入目标端失败）生效，而ConnectException属于连接器级别的致命异常——框架会将这类异常判定为连接器无法继续运行的严重错误，直接触发停止，不会进入错误容忍流程。

具体解决方法 #

• 捕获记录级异常并上报：在SinkTask.put()方法中，针对单条消息的处理失败，捕获具体业务/IO异常，调用context.raiseError(record, e)将错误上报给框架，而非直接抛出ConnectException。只有当遇到全局不可恢复错误（比如数据库连接池耗尽、核心配置无效）时，才抛出ConnectException。
• 实现错误处理相关接口：如果需要自定义错误指标或更精细的错误处理逻辑，可以实现ErrorHandlingMetrics接口，让连接器更好地融入框架的错误处理体系。
• 完善配套配置：同时配置errors.deadletterqueue.topic.name等死信队列参数，确保errors.tolerance=all时，失败消息能被转发到DLQ，连接器持续运行。

代码示例 #

@Override
public void put(Collection<SinkRecord> records) {
    for (SinkRecord record : records) {
        try {
            // 自定义消息处理逻辑
            processRecord(record);
        } catch (IOException | DataFormatException e) {
            // 上报单条记录错误，触发框架错误容忍机制
            context.raiseError(record, e);
        } catch (RuntimeException e) {
            // 判定为全局致命错误时，才抛出ConnectException
            throw new ConnectException("全局不可恢复错误", e);
        }
    }
}

内容的提问来源于stack exchange，提问作者suraj shinde