Quarkus Kafka：Multi处理场景下指定Header传播失效问题

使用SmallRye Kafka实现了基于Multi的消息处理逻辑：接收Multi<Bytes>类型的输入消息，通过flatMap将单条消息拆分为多条字符串输出。已在输出连接器配置中指定propagate-headers: "HEADER1,HEADER2"，但仅OpenTelemetry Header能正常传播，自定义的HEADER1和HEADER2未生效。

核心原因 #

当直接接收Multi<Bytes>时，SmallRye Kafka仅传递消息体，原始消息的Header元数据不会被关联到flatMap生成的每个子消息中。而OpenTelemetry Header是通过Reactive Context自动传播的，因此不受此限制，但自定义Header需要手动处理。

解决步骤 #

1. 改用IncomingKafkaRecord接收完整消息 #

将输入参数从Multi<Bytes>改为Multi<IncomingKafkaRecord<K, V>>，这样可以获取原始消息的完整元数据（包括Header）。

2. 手动将原始Header复制到每个输出消息 #

在flatMap处理逻辑中，将拆分后的每个字符串包装为OutgoingKafkaRecord，并手动复制需要传播的Header。

代码示例：

import io.smallrye.reactive.messaging.kafka.IncomingKafkaRecord;
import io.smallrye.reactive.messaging.kafka.OutgoingKafkaRecord;
import org.apache.kafka.common.header.Header;
import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Outgoing;
import io.smallrye.mutiny.Multi;
import jakarta.validation.constraints.NotNull;
import org.apache.kafka.common.utils.Bytes;

import java.util.List;

@Incoming("generic-in")
@Outgoing("generic-out")
public Multi<OutgoingKafkaRecord<String, String>> consume(@NotNull Multi<IncomingKafkaRecord<String, Bytes>> messages) {
    return messages.flatMap(record -> {
        // 解析原始Bytes为多个字符串（替换为你的实际解析逻辑）
        List<String> parsedStrings = parseMessageIntoMultipleStrings(record.value());
        
        // 为每个拆分后的消息复制指定Header
        return Multi.createFrom().iterable(parsedStrings)
                .map(str -> {
                    OutgoingKafkaRecord<String, String> outgoingMsg = OutgoingKafkaRecord.of(str);
                    
                    // 复制HEADER1和HEADER2
                    record.headers().forEach(header -> {
                        String headerName = header.key();
                        if ("HEADER1".equals(headerName) || "HEADER2".equals(headerName)) {
                            outgoingMsg.addHeader(headerName, header.value());
                        }
                    });
                    
                    return outgoingMsg;
                });
    });
}

// 你的实际消息解析方法
private List<String> parseMessageIntoMultipleStrings(Bytes message) {
    // 替换为你的解析逻辑，将Bytes拆分为多个字符串
    return List.of("string1", "string2", "string3");
}

3. 保留原有配置（可选） #

输出连接器的propagate-headers配置可以保留，它不会影响手动复制Header的逻辑，同时能确保OpenTelemetry等自动传播的Header继续生效。

优化建议 #

如果需要传播的Header较多或需要从配置中动态读取，可以将Header名称列表注入到类中，避免硬编码：

import org.eclipse.microprofile.config.inject.ConfigProperty;
import jakarta.inject.Inject;

// 在类中注入配置的Header列表
@Inject
@ConfigProperty(name = "mp.messaging.outgoing.generic-out.propagate-headers")
List<String> propagateHeaders;

// 然后在复制Header时使用该列表
record.headers().forEach(header -> {
    if (propagateHeaders.contains(header.key())) {
        outgoingMsg.addHeader(header.key(), header.value());
    }
});

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