Lambda中Quarkus Kafka生产者：阻塞至消息确认的元数据发送方案

方法一：直接使用Quarkus集成的原生KafkaProducer #

Quarkus支持直接注入配置好的KafkaProducer实例，通过它可以发送包含完整元数据（topic、key、headers、分区等）的ProducerRecord，同时利用Future.get()阻塞主线程直到收到Broker的确认：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import jakarta.inject.Inject;

public class LambdaKafkaProducer {

    @Inject
    KafkaProducer<String, String> kafkaProducer; // 已通过application.properties配置参数的生产者

    public void sendWithFullMetadata() throws Exception {
        // 构造包含所有元数据的消息
        ProducerRecord<String, String> record = new ProducerRecord<>(
            "your-target-topic",
            "custom-message-key",
            "actual-payload-content",
            null, // 指定分区（可选）
            System.currentTimeMillis(), // 自定义时间戳
            null, // 可添加自定义headers
            null
        );

        // 发送并阻塞等待确认
        RecordMetadata metadata = kafkaProducer.send(record).get();
        // 可按需处理确认后的元数据
        System.out.println("消息已确认：分区" + metadata.partition() + "，偏移量" + metadata.offset());
    }
}

这种方式完全可控，能直接操作所有Kafka消息元数据，且通过阻塞调用确保Lambda在消息发送完成后再终止，完美适配Lambda的执行模型。

方法二：扩展MicroProfile Emitter，封装带元数据的消息 #

如果偏好使用MicroProfile Reactive Messaging的Emitter，可以通过自定义DTO封装元数据和负载，再在消息处理环节解析并构造完整Kafka消息：

1. 定义包含元数据的DTO：

public class KafkaMsgWithMeta<T> {
    private String topic;
    private String key;
    private T payload;
    // 可按需添加headers、分区等字段
    // getter/setter省略
}

2. 使用Emitter发送该DTO并阻塞等待确认：

import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;
import jakarta.inject.Inject;
import java.util.concurrent.ExecutionException;

public class LambdaEmitterProducer {

    @Inject
    @Channel("kafka-out") // 对应配置文件中的输出通道
    Emitter<KafkaMsgWithMeta<String>> emitter;

    public void sendWithMetadata() throws ExecutionException, InterruptedException {
        KafkaMsgWithMeta<String> msg = new KafkaMsgWithMeta<>();
        msg.setTopic("your-target-topic");
        msg.setKey("msg-key-001");
        msg.setPayload("hello from lambda");

        // 发送并阻塞等待完成
        emitter.send(msg).toCompletableFuture().get();
    }
}

3. 编写消息处理器解析DTO并构造Kafka消息：

import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Outgoing;
import org.apache.kafka.clients.producer.ProducerRecord;

public class KafkaMsgProcessor {

    @Incoming("kafka-out")
    @Outgoing("kafka-out-processed")
    public ProducerRecord<String, String> processMsg(KafkaMsgWithMeta<String> msg) {
        return new ProducerRecord<>(
            msg.getTopic(),
            msg.getKey(),
            msg.getPayload()
        );
    }
}

最后在application.properties中配置处理后的输出通道：

mp.messaging.outgoing.kafka-out-processed.connector=smallrye-kafka
mp.messaging.outgoing.kafka-out-processed.topic=${kafka.target.topic}
# 其他Kafka生产者配置（如bootstrap.servers等）

关键注意事项 #

• 务必设置足够长的Lambda执行超时时间，避免消息未发送完成就被强制终止。
• 处理发送过程中的异常（如Broker不可用、网络中断），避免Lambda静默失败。
• 使用原生KafkaProducer时，Quarkus会自动管理实例生命周期，无需手动关闭。

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