使用StreamBridge发送CloudEvent至Kafka时自定义Header被覆盖/丢失的解决方法咨询

我目前在使用Spring Cloud Stream 2022.0.0搭配Spring Boot 3.0.1，通过StreamBridge向Kafka主题发送CloudEvent消息。我用CloudEventMessageBuilder构建消息和自定义Header，但发现发送到Kafka后，这些Header要么被覆盖要么直接丢失了，具体表现如下：

我使用的核心代码如下：

Message<String> message = CloudEventMessageBuilder.withData(payload)
    .setType("CustomType")
    .setSource("CustomSource")
    .setSubject("CustomSubject")
    .build();

streamBridge.send("channel-out-0", message);

虽然消息能成功发送到Kafka主题，但自定义的CloudEvent Header完全没有按预期传递，想请教一下各位大佬，应该怎么配置或者调整代码才能让这些自定义Header正确传递到Kafka呢？

可能的解决方向： #

这种情况一般是Spring Cloud Stream默认的CloudEvent处理逻辑或绑定配置导致的，你可以尝试以下几种方案排查解决：

• 调整CloudEvent绑定配置：在配置文件（application.yml/properties）中，针对输出通道配置Header传递策略，强制保留指定Header：

  spring:
    cloud:
      stream:
        bindings:
          channel-out-0:
            producer:
              header-mode: headers
              headers: ce_type,ce_source,ce_subject
  

  同时可以检查是否开启了原生编码，设置use-native-encoding: false避免Header被序列化逻辑干扰。

• 直接构建CloudEvent对象发送：绕过CloudEventMessageBuilder，直接使用CloudEvent原生API构建对象再发送，这样能让框架更准确地识别CloudEvent属性：

  import io.cloudevents.CloudEvent;
  import io.cloudevents.core.builder.CloudEventBuilder;
  import java.net.URI;
  import java.util.UUID;
  
  // 构建CloudEvent对象
  CloudEvent cloudEvent = CloudEventBuilder.v1()
      .withId(UUID.randomUUID().toString())
      .withType("CustomType")
      .withSource(URI.create("CustomSource"))
      .withSubject("CustomSubject")
      .withData(payload.getBytes())
      .build();
  
  // 通过StreamBridge发送
  streamBridge.send("channel-out-0", cloudEvent);
  

• 自定义消息转换器：如果默认的CloudEventMessageConverter存在Header覆盖逻辑，可以自定义转换器并注入到Spring容器中，确保自定义的CloudEvent属性被正确映射到Kafka Header：

  @Bean
  public CloudEventMessageConverter cloudEventMessageConverter() {
      return new CloudEventMessageConverter() {
          @Override
          protected Message<?> convertFromCloudEvent(CloudEvent cloudEvent, MimeType outputMimeType, ConversionService conversionService) {
              // 自定义转换逻辑，保留所有CloudEvent属性
              return super.convertFromCloudEvent(cloudEvent, outputMimeType, conversionService);
          }
      };
  }
  

• 检查Kafka生产者配置：确认Kafka生产者的Header序列化器配置是否正确，避免Header内容被意外修改，比如设置：

  spring:
    cloud:
      stream:
        kafka:
          bindings:
            channel-out-0:
              producer:
                configuration:
                  header.serializer: org.apache.kafka.common.serialization.StringSerializer
  

备注：内容来源于stack exchange，提问作者Moazzam Khan