Spring Kafka批量监听器自定义Header获取异常，需配置默认值为0

DevOps 小满

2026-6-20

问题分析

你当前的问题是Spring Kafka批量监听器中，cushion_next_retry_attempt Header在缺失时始终返回null，没有按预期获取到默认值0。核心原因是批量转换器的HeaderMapper配置冲突，以及监听器中未对列表内的null元素做兜底处理。

解决方案

1. 修正批量转换器配置

BatchMessagingMessageConverter无需单独设置HeaderMapper，它会委托内部的RecordMessageConverter处理每条消息的头部映射。移除批量转换器的HeaderMapper配置，避免覆盖单条消息的头部处理逻辑：

@Bean
public BatchMessagingMessageConverter batchConverter() {
    return new BatchMessagingMessageConverter(converter());
    // 删除此行：batchMessagingMessageConverter.setHeaderMapper(new CustomKafkaHeaderMapper());
}

2. 确保自定义HeaderMapper生效

你的CustomKafkaHeaderMapper逻辑本身是正确的，它会在每条消息转换为Spring Message时，为缺失的cushion_next_retry_attempt Header添加默认值0。只需保证converter()方法返回的JsonMessageConverter正确绑定了该Mapper：

@Bean
public RecordMessageConverter converter() {
    JsonMessageConverter jsonMessageConverter = new JsonMessageConverter();
    jsonMessageConverter.setHeaderMapper(new CustomKafkaHeaderMapper());
    return jsonMessageConverter;
}

3. 监听器内处理null值兜底

即使HeaderMapper正常工作，也建议在代码中对列表内的null元素做兜底处理，避免空指针异常：

@KafkaListener(id = "${spring.kafka.listener.cancel-auth-linkage.id}",
        topics = "${spring.kafka.listener.cancel-auth-linkage.topic.linkage}", autoStartup = "false",
        batch = "true",
        groupId = "cushion",
        concurrency = "4")
@Transactional("kafkaTransactionManager")
public void listen(List<CancelAuthorizationLinkage> messages,
                   @Header(KafkaHeaders.RECEIVED_KEY) List<String> keys,
                   @Header(
                           value = "cushion_next_retry_attempt",
                           required = false
                   )
                   List<Integer> nextRetryAttempts) throws InvalidValueException {
    for (int i = 0; i < messages.size(); i++) {
        CancelAuthorizationLinkage message = messages.get(i);
        String key = keys.get(i);
        // 对单个元素的null做兜底，默认设为0
        int nextRetryAttempt = (nextRetryAttempts != null && nextRetryAttempts.get(i) != null) 
                               ? nextRetryAttempts.get(i) 
                               : 0;
        try {
            processor.process(message);
        } catch (Exception e) {
            CompletableFuture<SendResult<String, CancelAuthorizationLinkage>> handle =
                    retryHandler.handle(message, listenerPropertiesService, e, nextRetryAttempt, key);
        }
    }
}

4. 可选：@Header的defaultValue补充（仅适用于列表整体为null的场景）

@Header的defaultValue仅在整个列表为null时生效，无法处理列表内单个元素为null的情况，可作为补充配置：

@Header(
        value = "cushion_next_retry_attempt",
        required = false,
        defaultValue = "0"
)
List<Integer> nextRetryAttempts

验证步骤

调试CustomKafkaHeaderMapper.toHeaders方法，确认消费无该Header的消息时，是否执行并添加了默认值0
检查监听器中nextRetryAttempts列表的元素，确认缺失Header的位置是否被正确处理为0

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