有没有办法阻止Kafka消费者在一个特定的偏移？

Question

例如，分区有1-10的偏移量。我只想从3-8消费。在消耗了第8条消息后，程序应该退出。

Answer 1

是的，kafka没有提供这个功能，但是您可以在您的消费者代码中实现这个功能。您可以尝试使用commitsync()来控制这一点。

公共无效commitSync（映射偏移）

为指定的主题和分区列表提交指定的偏移量。这就抵消了Kafka。使用该API提交的偏移量将在每次重新平衡后的第一次提取中使用，也将在启动时使用。因此，如果您需要在Kafka以外的任何地方存储偏移量，则不应使用此API。提交的偏移量应该是应用程序将要使用的下一个消息，即lastProcessedMessageOffset+1。

 while (goAhead) {
     ConsumerRecords<String, String> records = consumer.poll(100);
     for (ConsumerRecord<String, String> record : records) {
         if (record.offset() > OFFSET_BOUND) {
            consumer.commitSync(Collections.singletonMap(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset())));
            goAhead = false;
            break;           
         }
         process(record);
     }
 }