Kafka Streams默认反序列化异常处理器触发场景及自定义反序列化器疑问

问题描述 #

我的Kafka StreamConfig配置如下：

Properties properties = new Properties();
properties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG , LogAndContinueExceptionHandler.class);
...

我实现了自定义反序列化器：

public class KeyDeserializer implements Deserializer<Key>

其deserialize方法定义如下：

@Override
public Key deserialize(String s, byte[] bytes)

我的疑问是：该方法中反序列化时发生的任何RuntimeException都会被LogAndContinueExceptionHandler捕获，还是仅捕获Kafka特定异常？我未在文档中找到相关说明，需确保反序列化时无论发生何种异常，流都能记录日志并继续运行。

解答 #

你的假设是正确的——自定义反序列化器deserialize方法中抛出的所有RuntimeException都会被LogAndContinueExceptionHandler捕获，并非仅局限于Kafka特定异常。

当Kafka Streams调用反序列化器的deserialize方法时，会将该方法抛出的任何运行时异常统一包装为DeserializationException，然后传递给你配置的异常处理器。LogAndContinueExceptionHandler的核心逻辑就是记录异常日志，并返回CONTINUE处理结果，确保流任务不会因为单个消息的反序列化失败而中断，完全符合你想要的“记录日志并继续运行”的需求。

内容的提问来源于stack exchange，提问作者Vytautas Šerėnas