Kafka消费品属性从一开始就是一个话题

Question

我试图写一个Kafka消费者从一开始就消费这些信息。我可以从控制台消费者开始使用同样的方法

但是我在JAVA API中找不到相应的属性。

 def consumeFromKafka(topic: String) = {
    val props = new Properties()
    props.put("bootstrap.servers", "localhost:9092")
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    props.put("auto.offset.reset", "latest")
    props.put("group.id", "consumer-group")
    val consumer: KafkaConsumer[String, String] = new KafkaConsumer[String, String](props)
    consumer.subscribe(util.Arrays.asList(topic))
    while (true) {
      val record = consumer.poll(1000).asScala
      for (data <- record.iterator)
        println(data.value())
    }
  }

还有一个问题是什么应该是价值。Avro消息的反序列化程序？

Answer 1

可通过设置auto来实现Kafka控制台消费者中使用的影响--从一开始。抵消将重置为最早的。与独特的/新的组组合使用。id它具有相同的效果。

基本上，您想要创建一个新的消费者组（通过group.id），由于Kafka Broker不知道这个消费者组，它会根据配置自动重置这个消费者组的偏移量auto.offset.reset.当设置为最早时，它将从头开始。当设置为最新的时，它等待新的传入数据。

关于Avro反序列化，这里可能会有所帮助。