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

Question

我需要打印/日志/存储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);
}

Answer 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