如何在Kafka中处理JSON/文本多格式消息并避免偏移量阻塞？

方案1：消费者端异常捕获+强制偏移提交 #

在AVRO反序列化逻辑外层添加异常捕获，当遇到非AVRO格式消息触发SerializationException时，直接提交该消息的偏移量（+1），跳过坏消息，避免阻塞后续消费。

代码示例（Java Kafka客户端）：

ConsumerRecords<String, GenericRecord> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, GenericRecord> record : records) {
    try {
        // 正常AVRO消息处理逻辑
        processAvroRecord(record.value());
        // 手动提交当前偏移（仅适用于手动提交模式）
        consumer.commitSync(Collections.singletonMap(
            record.topicPartition(), 
            new OffsetAndMetadata(record.offset() + 1)
        ));
    } catch (SerializationException e) {
        // 捕获反序列化失败，判定为非AVRO格式消息
        System.err.println("跳过无效格式消息，offset: " + record.offset());
        // 强制提交偏移，跳过当前坏消息
        consumer.commitSync(Collections.singletonMap(
            record.topicPartition(), 
            new OffsetAndMetadata(record.offset() + 1)
        ));
    }
}

注意：需确保消费者配置enable.auto.commit为false，使用手动提交模式精准控制偏移。

方案2：死信队列（DLQ）转发坏消息 #

将无法反序列化的消息转发至专门的死信主题，既保留坏消息用于后续排查，又不阻塞主主题的消费流程。

代码示例：

// 初始化死信队列生产者
Producer<String, byte[]> dlqProducer = new KafkaProducer<>(getDlqProducerConfigs());

ConsumerRecords<String, GenericRecord> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, GenericRecord> record : records) {
    try {
        processAvroRecord(record.value());
        consumer.commitSync(Collections.singletonMap(
            record.topicPartition(), 
            new OffsetAndMetadata(record.offset() + 1)
        ));
    } catch (SerializationException e) {
        // 将原消息（字节数组）转发至死信队列
        dlqProducer.send(new ProducerRecord<>(
            "your-topic-dlq", 
            record.key(), 
            record.value()
        ));
        dlqProducer.flush();
        // 提交偏移，跳过坏消息
        consumer.commitSync(Collections.singletonMap(
            record.topicPartition(), 
            new OffsetAndMetadata(record.offset() + 1)
        ));
    }
}

方案3：自定义多格式兼容反序列化器 #

实现一个兼容AVRO、JSON、纯文本的自定义反序列化器，自动适配不同格式的消息，从根源避免反序列化失败导致的阻塞。

代码示例：

public class MultiFormatDeserializer implements Deserializer<Object> {
    private final AvroDeserializer avroDeserializer = new AvroDeserializer();
    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        avroDeserializer.configure(configs, isKey);
    }

    @Override
    public Object deserialize(String topic, byte[] data) {
        try {
            // 优先尝试AVRO反序列化
            return avroDeserializer.deserialize(topic, data);
        } catch (SerializationException avroEx) {
            try {
                // AVRO失败，尝试JSON反序列化
                return objectMapper.readValue(data, Object.class);
            } catch (JsonProcessingException jsonEx) {
                // JSON失败，按UTF-8字符串返回
                return new String(data, StandardCharsets.UTF_8);
            }
        }
    }

    @Override
    public void close() {
        avroDeserializer.close();
    }
}

使用时，将消费者配置value.deserializer设为该自定义类的全限定名。

方案4：上游生产者端格式校验（源头控制） #

如果消息来自内部生产系统，可在生产者端添加拦截器，校验消息格式，不符合要求的消息直接拦截或转换为兼容格式，避免坏消息流入Kafka主题。

拦截器示例：

public class FormatValidationInterceptor implements ProducerInterceptor<String, Object> {
    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public ProducerRecord<String, Object> onSend(ProducerRecord<String, Object> record) {
        try {
            // 校验是否为合法AVRO或JSON格式（根据业务需求调整逻辑）
            if (record.value() instanceof GenericRecord) {
                return record;
            }
            objectMapper.writeValueAsBytes(record.value());
            return record;
        } catch (JsonProcessingException e) {
            // 转换为兼容字符串格式或直接抛出异常拦截消息
            return new ProducerRecord<>(record.topic(), record.key(), record.value().toString());
        }
    }

    // 实现其他接口方法（略）
}

内容的提问来源于stack exchange，提问作者Ashok Kumar Singh