即使使用 auto.offset.reset，我的 Java 消费者也无法从代理读取消息 - 最早

Question

我的代理中有一个名为“测试”的主题。我用 CLI 检查了它。

我创建了一个 java 生产者来向主题 test 发送消息。我可以从我的 CLI 中使用它们。

.\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning（我在 Windows 上运行）

但是，当我在 Java Consumer 程序中运行它时，即使我将 auto.offset.reset 设置为 earliest，它也不会使用任何消息。我究竟做错了什么？

public class Consumer1 {

    public static void main(String[] args) {

        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "jin");
        props.put("enable.auto.commit", "true");
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        //consumer.subscribe(Collections.singletonList("test"));
        consumer.subscribe(Arrays.asList("test"));

        try {
            while (true) {              
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                //ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records){
                    System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());
                }
                //consumer.commitAsync();               
            }
        } catch (Exception e){
            e.printStackTrace();
        } finally {
            consumer.close();
            System.out.println("closed");
        }   
    }
}

Answer 1

auto.offset.reset 如果是全新的消费者组，或者消费者组的偏移量被删除，该属性将会生效。它不适用于已经在 Kafka 中存储偏移量的消费者组

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