在没有提交操作的情况下使用 Java Kafka Consumer 是否正确？答案

【问题标题】：Is it correct to use Java Kafka Consumer without commit operation?在没有提交操作的情况下使用 Java Kafka Consumer 是否正确？
【发布时间】：2019-08-11 12:23:25
【问题描述】：

我需要读取从开始偏移到结束偏移的记录集。为此，我使用专用的 Kafka 消费者。我至少可以接受一次语义（如果给定的应用程序实例出现故障，并且新的应用程序实例从该起始偏移量重新读取记录）。

那么，我可以使用这样的代码吗？

private static KafkaConsumer<Long, String> createConsumer() {

    final Properties props = new Properties();

    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName());
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    return new KafkaConsumer<>(props);
}

public void process() {

    KafkaConsumer consumer = createConsumer();
    TopicPartition topicPartition = new TopicPartition("topic", 2);
    consumer.assign(List.of(topicPartition));

    long startOffset = 42;
    long endOffset = 100;

    consumer.seek(topicPartition, startOffset);

    boolean isRunning = true;
    while (isRunning) {
        final ConsumerRecords<Long, String> consumerRecords = consumer.poll(1000);

        for (ConsumerRecord<Long, String> record : consumerRecords) {
            if (record.offset() >= endOffset) {
                isRunning = false;
                break;
            }
        }
    }

    consumer.close();
}

所以：

我没有commit()
我禁用auto-commit
我没有group-id

它是正确的代码吗？还是有一些隐藏的问题？

【问题讨论】：

你可以，尽管看起来你有错误的技术。将这些事件加载到适当的数据库中并按需重新加载会有什么问题？这个答案也可以是interesting。有几种方法可以实现您想做的事情
How to read data using Kafka Consumer API from beginning?的可能重复
您还应该更改您的代码以使用props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
@senseiwu 比props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); 更快吗？
如果不添加，则不会从最早的offset开始消费给新的consumer。

标签： java apache-kafka kafka-consumer-api

【解决方案1】：

是的，这是正确的用法，您不应该遇到任何问题。这不是 Kafka 消费者的典型用法，但它是允许的。

来自官方KafkaConsumerjavadoc（我的亮点）：

https://kafka.apache.org/21/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html

控制消费者的位置

在大多数用例中，消费者将简单地从头到尾消费记录，定期提交其位置（自动或手动）。 然而 Kafka 允许消费者手动控制其位置，在分区中随意向前或向后移动。这意味着消费者可以重新消费旧记录，或跳到最近的记录而不实际使用中间记录。在某些情况下，手动控制消费者的位置很有用。

...

Kafka 允许使用 seek(TopicPartition, long) 指定位置来指定新位置。还可以使用特殊方法来查找服务器维护的最早和最新偏移量（分别为 seekToBeginning(Collection) 和 seekToEnd(Collection)）。

【讨论】：