基于Micronaut Kafka实现Exactly Once语义及修改消息Key的方案问询

问题1：使用@SendTo时同时修改消息键 #

你提到的@SendTo默认沿用原消息键的问题，可以通过返回ProducerRecord对象解决——这个对象允许你同时指定新的键、值和目标主题，且依然能让偏移量提交纳入生产者事务，保证Exactly Once语义。

修改后的代码示例：

@KafkaListener(
    groupId = "mx-group",
    offsetReset = EARLIEST,
    offsetStrategy = SEND_TO_TRANSACTION,
    producerClientId = "prod-client-id",
    producerTransactionalId = "prod-client-tx"
)
class MyListener {
    
    @Topic("source.topic")
    @SendTo
    fun handle(
        @KafkaKey key: SourceKey,
        @MessageBody value: SourceValue
    ): ProducerRecord<TargetKey, TargetValue> {
        val newKey = convertKey(key)
        val newValue = convertValue(value)
        return ProducerRecord("target.topic", newKey, newValue)
    }

    // 省略convertKey和convertValue实现
}

注：@SendTo可以不指定主题，直接从ProducerRecord中读取；也可以保留@SendTo("target.topic")，两者均生效。返回ProducerRecord后，Micronaut Kafka会自动发送该记录到指定主题，同时将消费者偏移量提交绑定到生产者事务，确保事务提交时才提交偏移量，实现Exactly Once。

问题2：使用@KafkaClient时将偏移量提交纳入事务 #

若选择用@KafkaClient注入生产者，需将监听器的offsetStrategy设为MANUAL，通过Ack对象手动将偏移量提交到事务中，同时确保生产者配置为事务性。

修改后的代码示例：

@KafkaListener(
    groupId = "my-group",
    offsetReset = EARLIEST,
    offsetStrategy = OffsetStrategy.MANUAL
)
class MyListener(private val producer: MyProducer) {
    
    @Topic("source.topic")
    fun handle(
        @KafkaKey key: SourceKey,
        @MessageBody value: SourceValue,
        ack: Ack // 注入Ack对象用于手动提交偏移量
    ) {
        LOGGER.info("Received record key {}, value {}", key, value)
        // 发送消息到目标主题（生产者已配置事务）
        producer.send(convertKey(key), convertValue(value))
        // 将偏移量提交到当前事务中
        ack.acknowledge()
    }

    // 省略convertKey、convertValue实现和LOGGER定义
}

@KafkaClient(
    id = "producer-id",
    transactionalId = "producer-tx",
    transactional = true // 明确开启生产者事务
)
interface MyProducer {
    @Topic("target.topic")
    fun send(@KafkaKey key: TargetKey, @MessageBody value: TargetValue)
}

关键说明： #

• 监听器offsetStrategy设为MANUAL，禁止Micronaut自动提交偏移量，由业务逻辑手动控制。
• 注入Ack对象并调用acknowledge()时，Micronaut会将偏移量提交请求加入当前生产者事务（因生产者已配置事务性）。
• @KafkaClient必须配置transactionalId并设置transactional = true，确保生产者在事务上下文内工作。
• 消息发送与偏移量提交会在同一事务中完成，要么全部成功，要么全部回滚，保证Exactly Once语义。

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