手動承認

この例は、コンシューマーアプリケーションでオフセットを手動で確認する方法を示しています。

この例では、spring.cloud.stream.kafka.bindings.input.consumer.ackMode を MANUAL に設定する必要があります。例として、対応する入力チャネル名を使用してください。

@SpringBootApplication
public class ManuallyAcknowdledgingConsumer {

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

 @Bean
 public Consumer<Message<?>> process() {
    return message -> {
        Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
        if (acknowledgment != null) {
         System.out.println("Acknowledgment provided");
         acknowledgment.acknowledge();
        }
    };
}