如何仅使用 Kafka 主题中的最新偏移量

Question

我正在开发一个使用 kafka 的 scala 应用程序。我的kafka消费者代码如下。

def getValues(topic: String): String  = {
        
  val props = new Properties()
  props.put("group.id", "test")
  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", "earliest")
  val consumer: KafkaConsumer[String, String] = new KafkaConsumer[String, String](props)

  val topicPartition = new TopicPartition(topic, 0)
  consumer.assign(util.Collections.singletonList(topicPartition))
  val offset = consumer.position(topicPartition) - 1
  val record = consumer.poll(Duration.ofMillis(500)).asScala
  for (data <- record)
    if(data.offset() == offset) val value = data.value()
  return value
}

在此我只想返回最新值。当我运行我的应用程序时，我得到以下日志：

 Resetting offset for partition topic-0 to offset 0

因为 val offset = consumer.position(topicPartition) - 1 变为 -1 并且 data.offset() 给出了所有偏移量的列表。结果我没有得到最新的价值。为什么它会自动将偏移量重置为0？我该如何纠正？我的代码有什么错误？或任何其他方式我可以从最新的偏移量中获取值？

Answer 1

您正在寻找 seek 方法 - 根据 JavaDocs - “覆盖消费者将在下一次轮询（超时）时使用的获取偏移量”。

还要确保您正在设置

props.put("auto.offset.reset", "latest")

对您的代码进行这两项修改后，以下内容对我来说仅获取所选主题中部分0 的最新偏移量的value：

import java.time.Duration
import java.util.Properties
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.common.TopicPartition
import collection.JavaConverters._

def getValues(topic: String): String  = {
    val props = new Properties()
    props.put("group.id", "test")
    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")
    val consumer: KafkaConsumer[String, String] = new KafkaConsumer[String, String](props)

    val topicPartition = new TopicPartition(topic, 0)
    consumer.assign(java.util.Collections.singletonList(topicPartition))
    val offset = consumer.position(topicPartition) - 1
    consumer.seek(topicPartition, offset)
    val record = consumer.poll(Duration.ofMillis(500)).asScala
    for (data <- record) {
      val value: String = data.value() // you are only reading one message if no new messages flow into the Kafka topic
    }
    value
}

Answer 2

在这一行，props.put("auto.offset.reset", "earliest")，您将 Kafka 消费者的参数 auto.offset.reset 设置为 earliest，这会将偏移量重置为最早。如果你想要最新的值，你应该使用latest。 你可以找到文档here。