如何移除Kafka生产消息中的x_datadog_kafka_produced头？

有三种可行的方案可以阻止这个头被注入，按推荐优先级排序：

1. 直接在Datadog Agent配置中禁用头注入 #

这是最彻底的方案，从监控端直接关闭该功能。找到Datadog Agent的Kafka集成配置文件（通常在/etc/datadog-agent/conf.d/kafka.d/conf.yaml），添加或修改以下配置：

init_config:
instances:
  - ... # 保留现有其他配置
    producer_inject_headers: false

如果是容器化部署，也可以通过环境变量直接设置：

DD_KAFKA_PRODUCER_INJECT_HEADERS=false

修改完成后重启Datadog Agent，即可停止往Kafka消息中注入该头。

2. 通过Spring Kafka拦截器移除该头 #

如果无法修改Datadog配置，可以在生产者端添加自定义拦截器，在消息发送前主动移除x_datadog_kafka_produced头：

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Map;

public class RemoveDatadogHeaderInterceptor implements ProducerInterceptor<Object, Object> {
    private static final String DATADOG_HEADER = "x_datadog_kafka_produced";

    @Override
    public ProducerRecord<Object, Object> onSend(ProducerRecord<Object, Object> record) {
        if (record.headers().lastHeader(DATADOG_HEADER) != null) {
            record.headers().remove(DATADOG_HEADER);
        }
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {}

    @Override
    public void close() {}

    @Override
    public void configure(Map<String, ?> configs) {}
}

然后在Spring Kafka的生产者配置中注册这个拦截器：

@Bean
public ProducerFactory<Object, Object> producerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    // 保留其他生产者配置
    configProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, 
        RemoveDatadogHeaderInterceptor.class.getName());
    return new DefaultKafkaProducerFactory<>(configProps);
}

3. 自定义HeaderMapper过滤该头 #

Spring Kafka的HeaderMapper负责处理消息头的序列化与反序列化逻辑，可以通过自定义实现过滤掉指定头：

import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
import java.util.Arrays;
import java.util.List;

public class FilteredKafkaHeaderMapper extends DefaultKafkaHeaderMapper {
    private static final List<String> EXCLUDED_HEADERS = Arrays.asList("x_datadog_kafka_produced");

    @Override
    protected boolean shouldMapOutbound(String headerName) {
        return !EXCLUDED_HEADERS.contains(headerName) && super.shouldMapOutbound(headerName);
    }
}

接着将这个自定义HeaderMapper配置到KafkaTemplate中：

@Bean
public KafkaTemplate<Object, Object> kafkaTemplate(ProducerFactory<Object, Object> producerFactory) {
    KafkaTemplate<Object, Object> template = new KafkaTemplate<>(producerFactory);
    template.setHeaderMapper(new FilteredKafkaHeaderMapper());
    return template;
}

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