使用 Reactor Kafka 消费指定范围消息后停止 Consumer

import org.apache.kafka.common.TopicPartition;
import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate;
import org.springframework.kafka.listener.ConsumerProperties;
import org.springframework.kafka.support.Acknowledgment;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.kafka.receiver.ReceiverOptions;
import reactor.kafka.receiver.ReceiverPartition;
import reactor.kafka.receiver.ReceiverRecord;

import java.time.Duration;
import java.util.Collections;
import java.util.Map;

public class KafkaConsumerExample {

    public Disposable consumeMessages(String topic, String groupId, String bootstrapServers) {
        TopicPartition topicPartition = new TopicPartition(topic, 0); // 假设只有一个 Partition

        // 配置 Consumer 属性
        Map<String, Object> consumerProps = Map.of(
                "bootstrap.servers", bootstrapServers,
                "group.id", groupId,
                "key.deserializer", org.apache.kafka.common.serialization.StringDeserializer.class,
                "value.deserializer", org.apache.kafka.common.serialization.StringDeserializer.class,
                "auto.offset.reset", "earliest" // 从最早的 Offset 开始消费
        );

        // 创建 ReceiverOptions
        ReceiverOptions<String, String> receiverOptions = ReceiverOptions.<String, String>create(consumerProps)
                .subscription(Collections.singleton(topic))
                .addAssignListener(partitions -> partitions.forEach(ReceiverPartition::seekToBeginning));

        // 创建 ReactiveKafkaConsumerTemplate
        ReactiveKafkaConsumerTemplate<String, String> kafkaConsumer = new ReactiveKafkaConsumerTemplate<>(receiverOptions);

        // 消费消息并停止 Consumer
        return kafkaConsumer
                .receive()
                .flatMap(record -> {
                    // 获取当前 Partition 的最新 Offset
                    Mono<Map<TopicPartition, Long>> endOffsetsMono = kafkaConsumer.doOnConsumer(consumer -> consumer.endOffsets(Collections.singleton(topicPartition)));

                    return endOffsetsMono.map(topicPartitionToLastOffset -> {
                        long lastOffset = topicPartitionToLastOffset.get(topicPartition);
                        return new RecordWithLastOffset(record, lastOffset);
                    });
                })
                .takeUntil(recordWithLastOffset -> recordWithLastOffset.record.offset() >= (recordWithLastOffset.lastOffset - 1))
                .subscribe(recordWithLastOffset -> {
                    ReceiverRecord<String, String> record = recordWithLastOffset.record;
                    Acknowledgment acknowledgment = record.receiverOffset();

                    System.out.printf("Received message: topic-partition=%s offset=%d key=%s value=%s\n",
                            acknowledgment.topicPartition(),
                            acknowledgment.offset(),
                            record.key(),
                            record.value());

                    acknowledgment.acknowledge();
                });
    }

    private static class RecordWithLastOffset {
        private final ReceiverRecord<String, String> record;
        private final long lastOffset;

        public RecordWithLastOffset(ReceiverRecord<String, String> record, long lastOffset) {
            this.record = record;
            this.lastOffset = lastOffset;
        }
    }


    public static void main(String[] args) {
        String topic = "your-topic-name";
        String groupId = "your-group-id";
        String bootstrapServers = "localhost:9092";

        KafkaConsumerExample example = new KafkaConsumerExample();
        Disposable disposable = example.consumeMessages(topic, groupId, bootstrapServers);

        // 保持程序运行一段时间，以便消费消息
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 取消订阅，停止消费
        disposable.dispose();
    }
}