获取使用 StreamBridge 处理 kafka 消息的分区和偏移量答案

【问题标题】：Get Partition and Offset number in which kafka message is being processed using StreamBridge获取使用 StreamBridge 处理 kafka 消息的分区和偏移量
【发布时间】：2021-03-02 09:17:26
【问题描述】：

我需要打印/记录/存储正在处理我的消息的 kafka 分区和偏移量。我怎样才能做到这一点？我正在使用 StreamBridge 发送来自生产者的消息，并且还使用功能性 spring kafka 流方法

Public delegateToSupplier(String id, Abc obj) {
Message<Abc> message = MessageBuilder.withPayload(obj).seHeaders(KafkaHeaders.MESSAGE_KEY, id.getBytes()).build();
streamBridge.send("out-topic", message);
}

【问题讨论】：

标签： spring spring-boot kafka-consumer-api apache-kafka-streams spring-kafka

【解决方案1】：

记录元数据可通过元数据通道（异步）获得：

@SpringBootApplication
public class So66436499Application {

    public static void main(String[] args) {
        SpringApplication.run(So66436499Application.class, args);
    }

    @Autowired
    StreamBridge bridge;

    @Bean
    public ApplicationRunner runner() {
        return args -> {
            this.bridge.send("myBinding", "test");
            Thread.sleep(5000);
        };
    }

    @ServiceActivator(inputChannel = "meta")
    void meta(Message<?> sent) {
        System.out.println("Sent: " + sent.getHeaders().get(KafkaHeaders.RECORD_METADATA, RecordMetadata.class));
    }

}

spring.cloud.stream.bindings.myBinding.destination=foo
spring.cloud.stream.kafka.bindings.myBinding.producer.record-metadata-channel=meta

Sent: foo-0@5

【讨论】：

我可以访问消费者的消息头吗？那就是我对kafka流使用spring boot的功能方法。 docs.spring.io/spring-cloud-stream-binder-kafka/docs/3.1.0/…。在 function,KStreamprocess() {} 中，有没有办法可以访问消息发送的标头？？
当它不相关时，总是问一个新问题。在 cmets 中就其他答案提出问题并不能帮助其他人寻找问题和答案。您可以访问连接到您的KStream 拓扑的自定义Processor 或Transformer 中的标头。如果您需要帮助，请提出一个新问题，显示您当前的消费者 bean。