为何KTable中LogAndContinueExceptionHandler未被调用？

核心原因 #

你的问题出在异常发生的阶段与KTable的状态特性不匹配：

• 无效JSON的校验异常发生在Serde反序列化环节，而KTable构建物化视图时，会先从源Topic加载数据初始化状态存储（或从changelog恢复状态）。这个阶段的反序列化异常属于流任务启动/恢复阶段的致命异常，LogAndContinueExceptionHandler只负责处理拓扑运行时（比如map、peek等操作）的业务异常，无法拦截Serde层面的初始化异常，因此直接导致任务从PENDING_ERROR转为ERROR。
• KStream能正常工作是因为它是无状态（或轻状态）的逐条消息处理，反序列化异常发生在单条消息的处理流程中，属于运行时异常，会被LogAndContinueExceptionHandler捕获并处理，所以能跳过无效消息继续运行。

解决方案 #

方案1：用ErrorHandlingDeserializer包装自定义Serde #

将你的SerdeUtil.pojoSerde()用ErrorHandlingDeserializer包装，它会把反序列化异常封装为可被异常处理器识别的形式，让KTable的状态加载阶段也能触发LogAndContinueExceptionHandler：

Serde<Pojo> pojoSerde = SerdeUtil.pojoSerde();
ErrorHandlingDeserializer<Pojo> errorHandlingDeserializer = new ErrorHandlingDeserializer<>(pojoSerde.deserializer());
Serde<Pojo> wrappedSerde = Serdes.serdeFrom(pojoSerde.serializer(), errorHandlingDeserializer);

之后在Materialized配置中使用这个包装后的Serde即可。

方案2：自定义Serde在反序列化时捕获异常 #

修改你的Pojo反序列化逻辑，在Serde内部捕获ConstraintViolationException，返回null或标记为无效的对象，再在拓扑中过滤掉无效值：

public class PojoDeserializer implements Deserializer<Pojo> {
    private final ObjectMapper objectMapper = new ObjectMapper();
    private final Validator validator = Validation.buildDefaultValidatorFactory().getValidator();

    @Override
    public Pojo deserialize(String topic, byte[] data) {
        if (data == null) return null;
        try {
            Pojo pojo = objectMapper.readValue(data, Pojo.class);
            Set<ConstraintViolation<Pojo>> violations = validator.validate(pojo);
            if (!violations.isEmpty()) {
                throw new ConstraintViolationException(violations);
            }
            return pojo;
        } catch (Exception e) {
            log.error("Failed to deserialize Pojo from topic: {}", topic, e);
            return null;
        }
    }
}

然后在拓扑中补充过滤逻辑：

.values
.filter((key, value) -> Objects.nonNull(key) && Objects.nonNull(value))
// 后续操作...

这种方式把异常拦截在Serde内部，避免触发任务级别的错误。

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