Kafka消费者遇不可信包异常，如何跳过对应消息？

你遇到的核心问题是**SerializationException在poll()方法内部就抛出了**，根本没进入后续的try-catch代码块，导致commitSync()从未执行，offset没有提交，所以无效消息会被反复拉取。

因为JsonDeserializer在反序列化不符合信任包的消息时，会在poll()拉取并解析消息的阶段直接抛出异常，你的代码还没拿到ConsumerRecords就报错退出了，自然无法提交offset跳过这条消息。

方案一：使用ErrorHandlingDeserializer包装反序列化器 #

Kafka提供的ErrorHandlingDeserializer可以将反序列化异常封装到消息的value中，避免poll()直接抛出异常，让你能在消费逻辑中处理无效消息并提交offset。

修改消费者配置 #

Map<String, Object> configurations = new HashMap<>();
configurations.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
configurations.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, "true");
configurations.put(ConsumerConfig.GROUP_ID_CONFIG, groupId );
configurations.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId );
// 用ErrorHandlingDeserializer包装值反序列化器
configurations.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
// 指定实际的反序列化器为JsonDeserializer
configurations.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName());
configurations.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
configurations.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
configurations.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, "52428800");
configurations.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, "52428800");
configurations.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "3600000");
configurations.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.RoundRobinAssignor");
configurations.put(JsonDeserializer.TRUSTED_PACKAGES, "com.example.proj1");
configurations.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1");
consumer = new KafkaConsumer<Object, Object>(configurations);
consumer.subscribe(topic);

修改消费逻辑 #

try {
    ConsumerRecords<Object, Object> records = consumer.poll(Duration.ofMillis(10000));
    if (!records.isEmpty()) {
        for (ConsumerRecord<Object, Object> record : records) {
            if (record.value() instanceof DeserializationException) {
                // 处理反序列化失败的消息，直接跳过
                LOGGER.error("无效包的消息，offset: {}", record.offset(), 
                    ((DeserializationException) record.value()).getCause());
            } else {
                // 处理正常消息
                handleMessages(Collections.singleton(record));
            }
        }
        // 无论是否有有效消息，都提交offset跳过所有拉取到的记录
        consumer.commitSync();
    }
} catch (Exception e) {
    LOGGER.error("消费异常", e);
}

方案二：自定义容错JsonDeserializer #

继承JsonDeserializer，重写deserialize方法捕获序列化异常，返回null标记无效消息，这样poll()不会抛出异常，你可以在消费逻辑中过滤掉null值的消息。

自定义反序列化器 #

public class TolerantJsonDeserializer<T> extends JsonDeserializer<T> {
    private static final Logger LOGGER = LoggerFactory.getLogger(TolerantJsonDeserializer.class);

    @Override
    public T deserialize(String topic, byte[] data) {
        try {
            return super.deserialize(topic, data);
        } catch (SerializationException e) {
            LOGGER.error("反序列化失败，topic: {}", topic, e);
            return null;
        }
    }
}

修改消费者配置 #

// 替换原有的JsonDeserializer为自定义的容错版本
configurations.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, TolerantJsonDeserializer.class);
// 其他配置保持不变
configurations.put(JsonDeserializer.TRUSTED_PACKAGES, "com.example.proj1");

修改消费逻辑 #

try {
    ConsumerRecords<Object, Object> records = consumer.poll(Duration.ofMillis(10000));
    if (!records.isEmpty()) {
        for (ConsumerRecord<Object, Object> record : records) {
            if (record.value() != null) {
                // 处理正常消息
                handleMessages(Collections.singleton(record));
            } else {
                // 跳过无效消息
                LOGGER.error("跳过无效消息，offset: {}", record.offset());
            }
        }
        // 提交offset跳过所有拉取的记录
        consumer.commitSync();
    }
} catch (Exception e) {
    LOGGER.error("消费异常", e);
}

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