Даже после установки «auto.offset.reset» в «последнее» получение ошибки OffsetOutOfRangeException

Question

я использую версию spark-sql-2.4.1 с Kafka 0.10 v.

Пока я пытаюсь использовать данные потребителем. выдает ошибку ниже даже после установки "auto.offset.reset" на "последний"

org.apache.kafka.clients.consumer.OffsetOutOfRangeException: Offsets out of range with no configured reset policy for partitions: {COMPANY_INBOUND-16=168}
    at org.apache.kafka.clients.consumer.internals.Fetcher.throwIfOffsetOutOfRange(Fetcher.java:348)
    at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:396)
    at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:999)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:937)
    at org.apache.spark.sql.kafka010.InternalKafkaConsumer.fetchData(KafkaDataConsumer.scala:470)
    at org.apache.spark.sql.kafka010.InternalKafkaConsumer.org$apache$spark$sql$kafka010$InternalKafkaConsumer$$fetchRecord(KafkaDataConsumer.scala:361)
    at org.apache.spark.sql.kafka010.InternalKafkaConsumer$$anonfun$get$1.apply(KafkaDataConsumer.scala:251)
    at org.apache.spark.sql.kafka010.InternalKafkaConsumer$$anonfun$get$1.apply(KafkaDataConsumer.scala:234)
    at org.apache.spark.util.UninterruptibleThread.runUninterruptibly(UninterruptibleThread.scala:77)
    at org.apache.spark.sql.kafka010.InternalKafkaConsumer.runUninterruptiblyIfPossible(KafkaDataConsumer.scala:209)
    at org.apache.spark.sql.kafka010.InternalKafkaConsumer.get(KafkaDataConsumer.scala:234)

в чем проблема? почему настройка не работает? Как это исправить?

Часть 2:

 .readStream()
                      .format("kafka")
                      .option("startingOffsets", "latest")
                      .option("enable.auto.commit", false)
                      .option("maxOffsetsPerTrigger", 1000)
                      .option("auto.offset.reset", "latest")
                      .option("failOnDataLoss", false)
                      .load();

Answer 1

auto.offset.reset игнорируется Spark Structured Streaming, вместо этого используйте опцию startOffsets

auto.offset.reset: установите исходную опцию startOffsets, чтобы указать, с чего начать. Структурированная потоковая передача управляет тем, какие смещения потребляются внутри, а не полагается на потребителя kafka. Это гарантирует, что никакие данные не будут пропущены, когда новые темы / разделы динамически подписаны. Обратите внимание на то, что initialOffsets применяется только при запуске нового потокового запроса, и это возобновление всегда будет выполняться с того места, где запрос был остановлен.

Источник