Spring Boot Kafka批量消费者：DefaultErrorHandler不记录错误且无法继续处理

为什么你的DefaultErrorHandler不生效？ #

你在parseKafkaEvent方法中自行捕获了所有解析异常并返回null，这些异常并没有抛到@KafkaListener方法之外。而DefaultErrorHandler只会处理listener方法抛出的未捕获异常，所以错误处理器根本没机会介入这些解析错误。

另外，在批量消费模式下，DefaultErrorHandler的默认行为是：一旦listener方法抛出异常，会对整个批量进行重试，而不是跳过单条失败记录。这也是你尝试添加恢复器后批量处理停止的原因——默认情况下，批量中只要有一条失败，整个批量都会被重试或终止。

实现“单条错误不中断批量+记录警告”的正确配置 #

1. 调整业务代码，让异常能被错误处理器捕获 #

移除parseKafkaEvent中的异常捕获，让解析失败的异常直接抛出，或者在逐个处理记录时抛出异常：

fun parseKafkaEvent(it: ConsumerRecord<String, String>): MyMappedEvent {
    return objectMapper.readValue(it.value()) as MyMappedEvent
}

@KafkaListener(
    topics = ["my-topic"],
    groupId = "my-group-id",
    batch = "true"
)
fun listen(records: ConsumerRecords<String, String>) {
    records.asSequence().forEach { record ->
        try {
            val event = parseKafkaEvent(record)
            processor.process(listOf(event)) // 改为单条处理，确保单条异常能抛出
        } catch (e: Exception) {
            // 不要自行捕获，让异常抛到listener外层交给错误处理器
            throw KafkaListenerException("处理记录失败", e, record)
        }
    }
}

2. 配置支持批量跳过的DefaultErrorHandler #

修改错误处理器配置，允许跳过失败记录并记录警告，同时绑定到容器工厂：

@Configuration
class KafkaConfiguration {
    @Bean
    fun errorHandler(): DefaultErrorHandler {
        val logger = LoggerFactory.getLogger("KafkaErrorHandler")
        // 定义恢复器：记录警告日志
        val recoverer = ConsumerRecordRecoverer { record, ex ->
            logger.warn("处理记录失败，已跳过。记录offset: ${record?.offset()}", ex)
        }
        val errorHandler = DefaultErrorHandler(recoverer, ExponentialBackOff()).apply {
            // 设置日志级别为WARN
            setLogLevel(KafkaException.Level.WARN)
            // 开启批量处理时跳过失败记录，继续处理后续记录
            setProcessBatchWhileFailed(true)
            // 添加无需重试的异常（如解析异常，重试无意义）
            addNotRetryableExceptions(JsonProcessingException::class.java, ClassCastException::class.java)
        }
        return errorHandler
    }

    // 配置容器工厂，绑定错误处理器并启用批量消费
    @Bean
    fun kafkaListenerContainerFactory(
        consumerFactory: ConsumerFactory<String, String>,
        errorHandler: DefaultErrorHandler
    ): ConcurrentKafkaListenerContainerFactory<String, String> {
        return ConcurrentKafkaListenerContainerFactory<String, String>().apply {
            setConsumerFactory(consumerFactory)
            isBatchListener = true
            setCommonErrorHandler(errorHandler)
            // 设置按单条记录提交offset，避免单条失败影响整个批量的offset提交
            containerProperties.ackMode = ContainerProperties.AckMode.RECORD
        }
    }
}

关键配置说明 #

• setProcessBatchWhileFailed(true)：开启后，批量处理中某条记录失败时，错误处理器会处理该记录（重试/恢复），然后继续处理批量中的下一条，不会中断整个批量。
• addNotRetryableExceptions：对解析失败这类无需重试的异常，直接标记为不可重试，避免无效重试消耗资源。
• AckMode.RECORD：确保每条记录处理完成后单独提交offset，避免单条失败导致整个批量的offset无法提交。

额外注意事项 #

• 如果必须保留批量处理逻辑（比如processor需要批量处理events），需要在processor内部捕获单条记录的异常，或者将批量拆分为单条处理并抛出异常，否则一旦processor抛出批量级异常，错误处理器还是会重试整个批量。
• 你的spring-kafka版本（2.9.12）支持setProcessBatchWhileFailed方法，该特性在2.8.0及以上版本可用。

内容的提问来源于stack exchange，提问作者Jens Brinkmann