添加delay头时Kafka生产者执行失败：Magic v1不支持记录头问题解决

一、解决Magic v1 does not support record headers报错 #

这个错误的原因是你的Kafka集群使用的是Magic v1版本的消息格式（对应Kafka 0.11.0.0之前的版本规范），该格式不支持消息头（Record Headers），而你在路由中给消息添加了delay头，导致Kafka生产者无法处理。

解决方法：

• 移除路由中的setHeader("delay", constant(delayMs))代码，因为这个Header对Kafka延迟投递没有任何作用，属于错误用法。
  修改后的代码：

  @Component
  public class kafkaProducerRoute extends RouteBuilder{
      @Override
      public void configure() throws Exception {
          int delayMs=10000;
  
          from("timer:time")
                  .setBody().constant("Hello, World!")
                  .to("kafka:hell1?brokers=localhost:9092")
                  .log("Message is produced");
      }
  }
  

• 如果业务确实需要使用消息头，需升级Kafka集群的消息格式：在Broker配置文件中设置message.format.version=2.0（或更高版本，比如当前稳定版的3.4），重启Broker后即可支持消息头。

二、实现Kafka延迟消息投递的正确方案 #

你的需求是通过Broker实现定时延迟投递，以下是两种可行方案：

1. 生产者端延迟发送（简单场景） #

如果不需要Broker端存储延迟消息，仅需生产者生成消息后延迟一段时间再发送给Broker，可使用Camel的delay()处理器实现：

@Component
public class kafkaProducerRoute extends RouteBuilder{
    @Override
    public void configure() throws Exception {
        int delayMs=10000;

        from("direct:sendDelayMsg") // 用direct替代timer，按需触发
                .setBody().constant("Hello, World!")
                .delay(delayMs) // 延迟指定毫秒后再发送
                .to("kafka:hell1?brokers=localhost:9092")
                .log("Delayed message is produced");
    }
}

这种方式适合简单延迟场景，但如果生产者宕机，未发送的延迟消息会丢失。

2. Broker端延迟投递（可靠场景） #

如果需要Broker端持久化延迟消息，到时间再投递给消费者，推荐使用Kafka的延迟队列方案，核心思路：

• 生产者将消息发送到一个「延迟主题」，消息中携带目标投递时间戳。
• 编写独立的消费者监听这个延迟主题，过滤出已到投递时间的消息，转发到业务主题。
• 业务消费者监听业务主题处理消息。

结合Camel实现的示例：

延迟消息生产者

@Component
public class DelayMsgProducer extends RouteBuilder{
    @Override
    public void configure() throws Exception {
        long delayMs=10000;
        long targetTime = System.currentTimeMillis() + delayMs;

        from("direct:sendDelayMsg")
                .setBody().constant("Hello, Delayed World!")
                .setHeader("targetDeliveryTime", constant(targetTime))
                .to("kafka:delay_topic?brokers=localhost:9092")
                .log("Delay message sent to delay topic");
    }
}

延迟消息转发器（Camel路由）

@Component
public class DelayMsgForwarder extends RouteBuilder{
    @Override
    public void configure() throws Exception {
        from("kafka:delay_topic?brokers=localhost:9092&groupId=delay_forward_group")
                .process(exchange -> {
                    long targetTime = exchange.getIn().getHeader("targetDeliveryTime", Long.class);
                    long currentTime = System.currentTimeMillis();
                    // 如果还没到投递时间，将消息重新发回延迟主题（可设置重试间隔）
                    if(currentTime < targetTime){
                        exchange.getIn().setHeader(KafkaConstants.KEY, exchange.getIn().getHeader(KafkaConstants.KEY));
                        exchange.getIn().setHeader(KafkaConstants.PARTITION, exchange.getIn().getHeader(KafkaConstants.PARTITION));
                        // 可添加重试次数限制，避免无限循环
                        exchange.setProperty(Exchange.ROUTE_STOP, true);
                        exchange.getContext().createProducerTemplate().send("kafka:delay_topic", exchange.getIn());
                    }
                })
                // 已到投递时间，转发到业务主题
                .to("kafka:hell1?brokers=localhost:9092")
                .log("Delayed message forwarded to business topic");
    }
}

这种方式基于Kafka的持久化特性，即使转发器宕机，重启后仍能继续处理延迟消息，可靠性更高。

另外，若使用Kafka 2.4+版本，也可考虑使用Kafka Streams的窗口功能实现延迟投递，原理类似，通过窗口等待指定时间后输出消息。

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