Spring云流Kafka：如何在生成Kafka主题的消息后访问Kafka头的recordMetadata

Question

我想在生成一条发送给Kafka主题的消息后，获取偏移量和分区信息。我通读了spring cloud stream kafka绑定文档，发现这可以通过fecting RECORD\u元数据kafka头来实现。

来自Spring文档：(https://cloud.spring.io/spring-cloud-static/spring-cloud-stream-binder-kafka/3.0.0.RELEASE/reference/html/spring-cloud-stream-binder-kafka.html#kafka-生产者财产）

RecordMetadataChannel成功发送结果应发送到的MessageChannel的bean名称；bean必须存在于应用程序上下文中。发送到通道的消息是已发送的消息（转换后，如果有的话），带有额外的标头KafkaHeaders。RECORD_METADATA。标头包含Kafka客户端提供的RecordMetadata对象；它包括记录在主题中写入的分区和偏移量。

结果元数据元数据=SendResultMg。getHeaders（）。get（KafkaHeaders.RECORD\u METADATA，RecordMetadata.class）

我已经在属性文件spring中将输出主题bean名称配置为消息通道。云流动Kafka。绑定。确认输出。制作人记录元数据通道=确认输出

我的自定义界面和生产者类如下：

public interface OutputAcknowledgement {    

    @Output("acknowledgement-out")
    MessageChannel output();

}

生产者类别：

@EnableBinding(OutputAcknowledgement.class)
public class AcknowledgementProducer {
   

    @Autowired
    OutputAcknowledgement outputAcknowledgement;

    public Boolean produce(Acknowledgement acknowledgement) {
        Message msg = MessageBuilder.withPayload(acknowledgement).build();

        boolean val = outputAcknowledgement.output().send(msg);

        RecordMetadata recordMetadata = msg.getHeaders().get(KafkaHeaders.RECORD_METADATA, RecordMetadata.class);

为记录获取null元数据。

请建议我的方法是否正确？

Answer 1

您获得null是因为它在您访问时不存在于该消息对象中。根据文档，元数据仅在成功发布后提供。请参阅此答案，了解如何通过为记录元数据通道提供处理程序/消费者来获取记录元数据。