如何将Kafka消息中的单个SmallObject放入死信队列（DLT）

要实现单个SmallObjects处理失败时仅将该对象投递到DLT，而非整个CreateMyObjectRequest消息，需要调整业务处理逻辑，单独捕获每个对象的异常并手动触发DLT投递，具体方案如下：

1. 封装DLT手动发布工具类 #

将DLT发布逻辑封装为可调用的组件，方便在单个对象处理失败时直接调用：

@Component
public class DltMessagePublisher {

    private final DeadLetterEnrichAndPublishRecoverer dltRecoverer;
    private final ObjectMapper objectMapper;

    public DltMessagePublisher(DeadLetterEnrichAndPublishRecoverer dltRecoverer, ObjectMapper objectMapper) {
        this.dltRecoverer = dltRecoverer;
        this.objectMapper = objectMapper;
    }

    public void publishSmallObjectToDlt(SmallObjects failedObject, String originalTopic, Exception exception) {
        try {
            // 将单个SmallObject序列化为字节数组，与原消息序列化方式保持一致
            byte[] payload = objectMapper.writeValueAsBytes(failedObject);
            // 构造模拟的ConsumerRecord，传递原主题信息用于DLT主题生成
            ConsumerRecord<byte[], byte[]> mockRecord = new ConsumerRecord<>(
                    originalTopic,
                    0, // 若需保留原消息分区，可通过@Header(KafkaHeaders.RECEIVED_PARTITION_ID)获取
                    System.currentTimeMillis(),
                    null, // 原消息有key的话可传入对应值
                    payload
            );
            // 调用DLT恢复器完成消息投递
            dltRecoverer.accept(mockRecord, exception);
        } catch (JsonProcessingException e) {
            // 记录序列化异常日志，避免影响后续对象处理
            e.printStackTrace();
        }
    }
}

2. 修改监听器与业务处理逻辑 #

拆分批量处理为单个对象处理，捕获每个SmallObjects的异常并触发DLT投递，不抛出全局异常导致整个消息重试：

@KafkaListener(topics = "#{__listener.createMyObjectConsumerProperties.topic.name}",
        groupId = "#{__listener.createMyObjectConsumerProperties.topic.consumerGroupId}",
        containerFactory = CREATE_MY_OBJECT_KAFKA_LISTENER_FACTORY)
public void consumeLead(CreateMyObjectRequest content, 
                        @Header(KafkaHeaders.RECEIVED_TOPIC) String originalTopic) {
    List<SmallObjects> smallObjectsList = content.getSmallObjectsList();
    if (smallObjectsList == null || smallObjectsList.isEmpty()) {
        return;
    }

    for (SmallObjects smallObj : smallObjectsList) {
        try {
            // 单独处理每个SmallObject
            injectedService.createSingleObject(smallObj);
        } catch (Exception e) {
            // 单个对象处理失败，投递至DLT
            dltMessagePublisher.publishSmallObjectToDlt(smallObj, originalTopic, e);
            // 记录失败日志，标记该对象处理状态
            System.err.printf("SmallObject处理失败已投递DLT: %s, 异常信息: %s%n", smallObj, e.getMessage());
        }
    }
}

同时调整业务服务，新增单个对象处理方法：

@Service
public class InjectedService {

    /**
     * 单个SmallObject的业务处理逻辑，失败时抛出异常
     */
    public void createSingleObject(SmallObjects smallObj) {
        // 这里实现单个对象的创建逻辑，如数据库写入、远程调用等
        // 处理失败时直接抛出异常，由上层捕获
    }

    // 保留原批量方法（若有其他场景使用）
    public void createObjects(CreateMyObjectRequest content) {
        // 原批量处理逻辑
    }
}

3. 可选：自定义单个对象的DLT主题 #

如果需要为单个SmallObjects设置独立的DLT主题（而非复用原消息的Error后缀主题），可修改工具类中的主题生成逻辑：

// 在publishSmallObjectToDlt方法中自定义DLT主题
String dltTopic = originalTopic + "-SmallObjectError";
// 直接构造目标DLT的ConsumerRecord
ConsumerRecord<byte[], byte[]> mockRecord = new ConsumerRecord<>(
        dltTopic,
        0,
        System.currentTimeMillis(),
        null,
        payload
);

或者调整原DeadLetterEnrichAndPublishRecoverer的主题解析器，根据消息类型动态生成主题：

@Bean
public DeadLetterPublishingRecoverer dltPublisherMyApp() {
    return new DeadLetterEnrichAndPublishRecoverer(getDltTemplate("127.0.0.1:9092"),
            (record, exception) -> {
                // 判断消息类型，生成对应DLT主题
                try {
                    Object payload = objectMapper.readValue(record.value(), Object.class);
                    if (payload instanceof SmallObjects) {
                        return new TopicPartition(record.topic() + "-SmallObjectError", record.partition());
                    } else {
                        return new TopicPartition(record.topic() + "Error", record.partition());
                    }
                } catch (JsonProcessingException e) {
                    // 解析失败时使用默认主题
                    return new TopicPartition(record.topic() + "Error", record.partition());
                }
            }, objectMapper);
}

关键注意事项 #

• 确保单个SmallObjects的序列化/反序列化方式与原消息一致，避免DLT消息无法解析。
• 单个对象处理失败后，原消息会被标记为消费成功，不会触发全局重试；若需对失败对象重试，需在DLT中单独配置重试逻辑。
• 完善失败日志记录，包含对象信息与异常栈，便于后续问题排查。

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