Невозможно прочитать тему Кафки из структурированной потоковой передачи Spark из-за несоответствия смещения

Question

Я пытаюсь прочитать записи из темы Kafka, используя приведенный ниже фрагмент кода:

ds_raw = (spark
                .readStream
                .format("kafka")
                .option("subscribe", GLOBAL_ARGS.kafka_topic)
                .option("kafka.bootstrap.servers", GLOBAL_ARGS.kafka_brokers)
                .option("failOnDataLoss", False)
                .option("startingOffsets", "latest") 
                .load())

и записать вывод в тему Kafka.

    sink_topic = "{0}_sink".format(GLOBAL_ARGS.kafka_topic)

    query = (df_output
            .selectExpr("CAST(id AS STRING) AS key", "to_json(struct(*)) AS value")
            .writeStream
            .format("kafka")
            .option("kafka.bootstrap.servers", GLOBAL_ARGS.kafka_brokers)
            .option("topic", sink_topic)
            .outputMode("update") 
            .trigger(processingTime=GLOBAL_ARGS.window_size)
            .option("checkpointLocation", "/checkpoint_tmp")
            .start())

Я получаю сообщение об ошибке ниже:

org.apache.spark.sql.streaming.StreamingQueryException: Writing job aborted.
=== Streaming Query ===
Identifier: [id = 811b7dff-aa61-4d3d-9784-394707180fb2, runId = 2fa6a3e4-1dc0-403a-bada-382237b61bd1]
Current Committed Offsets: {KafkaSource[Subscribe[topic_1]]: {"topic_1":{"2":0,"5":0,"4":0,"1":10552,"3":0,"0":0}}}
Current Available Offsets: {KafkaSource[Subscribe[topic_1]]: {"topic_1":{"2":0,"5":0,"4":0,"1":10565,"3":0,"0":0}}}

Кажется, что зафиксированное смещение и доступное смещение имеют разные значения.