多次调用消费者时如何重置Kafka消费者偏移量答案

【问题标题】：How to reset Kafka consumer offset when calling consumer many times多次调用消费者时如何重置Kafka消费者偏移量
【发布时间】：2021-01-06 07:51:08
【问题描述】：

我试图在调用消费者时重置消费者offset，这样当我多次调用消费者时，它仍然可以读取生产者发送的记录。我正在设置props.put("auto.offset.reset","earliest"); 并调用consumer.seekToBeginning(consumer.assignment()); 但是当我第二次调用消费者时它不会收到任何记录。我该如何解决这个问题？

   public ConsumerRecords<String, byte[]> consumer(){
    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("group.id", "test");
    //props.put("group.id", String.valueOf(System.currentTimeMillis()));
    props.put("auto.offset.reset","earliest");
    props.put("enable.auto.commit", "true");
    props.put("auto.commit.interval.ms", "1000");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
    KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(props);
    consumer.subscribe(Collections.singletonList("topiccc"));
    ConsumerRecords<String, byte[]> records = consumer.poll(100);
    consumer.seekToBeginning(consumer.assignment());
   /* List<byte[]> videoContents = new ArrayList<byte[]>();
    for (ConsumerRecord<String, byte[]> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
        videoContents.add(record.value());
    }*/

    return records;
} 

public String producer(@RequestParam("message") String message) {
    Map<String, Object> props = new HashMap<>();
    // list of host:port pairs used for establishing the initial connections to the Kakfa cluster
    props.put("bootstrap.servers", "localhost:9092");
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
    Producer<String, byte[]> producer = new KafkaProducer<>(props);
    Path path = Paths.get("C:/Programming Files/video-2012-07-05-02-29-27.mp4");
    ProducerRecord<String, byte[]> record = null;
    try {

        record = new ProducerRecord<>("topiccc", "keyyyyy"
                , Files.readAllBytes(path));

    } catch (IOException e) {
        e.printStackTrace();
    }
    producer.send(record);
    producer.close();
    //kafkaSender.send(record);

    return "Message sent to the Kafka Topic java_in_use_topic Successfully";
}

【问题讨论】：

如果您尝试手动操作，它会起作用吗？例如，kafka-consumer-groups.bat --bootstrap-server kafka-host:9092 --group my-group --reset-offsets --to-earliest --all-topics --execute
我需要在我的程序中完成
但我在线程“主”org.apache.kafka.common.config.ConfigException 中得到异常：尽管我已将其设置为“localhost:9092”，但在 bootstrap.servers 中没有给出可解析的引导 URL
您需要发布完整代码。
@GiorgosMyrianthous 我添加了代码。谢谢

标签： java apache-kafka

【解决方案1】：

在 Kafka Java 代码中，AUTO_OFFSET_RESET_CONFIG 上的 documentation 表示以下内容：

当 Kafka 中没有初始偏移量或服务器上不再存在当前偏移量时该怎么办（例如，因为该数据已被删除）：

最早：自动将偏移量重置为最早的偏移量

latest：自动将偏移量重置为最新的偏移量

无：如果没有为消费者组找到先前的偏移量，则向消费者抛出异常

其他：向消费者抛出异常。

这可以在 GitHub 中找到： https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

我们从他们的评论中可以看出，该设置仅在偏移量不在服务器上时使用。在问题中，偏移量是从服务器检索的，这就是为什么偏移量没有重置到开头而是停留在最后一个偏移量，看起来没有更多记录的原因。

您需要在服务器端显式重置偏移量，以按照问题中的要求解决此问题。

这是另一个描述如何做到这一点的答案。 https://stackoverflow.com/a/54492802/231860

这是一段允许我重置偏移量的代码。注意：如果调用 subscribe 方法，则不能调用 seekToBeginning。如果我自己使用 assign 方法分配分区，我只能让它工作。可惜。

// Create the consumer:
final Consumer<String, DataRecord> consumer = new KafkaConsumer<>(props);

// Get the partitions that exist for this topic:
List<PartitionInfo> partitions = consumer.partitionsFor(topic);

// Get the topic partition info for these partitions:
List<TopicPartition> topicPartitions = partitions.stream().map(info -> new TopicPartition(info.topic(), info.partition())).collect(Collectors.toList());

// Assign all the partitions to the topic so that we can seek to the beginning:
// NOTE: We can't use subscribe if we use assign, but we can't seek to the beginning if we use subscribe.
consumer.assign(topicPartitions);

// Make sure we seek to the beginning of the partitions:
consumer.seekToBeginning(topicPartitions);

是的，实现一个看似基本的用例似乎非常复杂。这可能表明整个 kafka 世界似乎只想读取一次流。

【讨论】：

【解决方案2】：

我通常会创建一个具有不同 group.id 的新消费者来再次读取记录。就这样吧：

props.put("group.id", Instant.now().getEpochSecond());

【讨论】：

【解决方案3】：

有一个解决方法（虽然不是生产解决方案），即每次使用时更改 group.id 配置值。在很多情况下，将auto.offset.reset 设置为earliest 是不够的。

【讨论】：

【解决方案4】：

当您希望一条消息被消费者多次消费时，理想的方法是创建具有不同消费者组的消费者，以便可以消费相同的消息。

但是如果你想让同一个消费者多次消费同一个消息，那么你可以使用 commit 和 offset 您将 auto.commit 设置得非常高或禁用它并按照您的逻辑提交

更多详情可以参考这里https://kafka.apache.org/10/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html 此 javadoc 提供了有关如何手动管理偏移量的详细信息

【讨论】：