Spark Structured Streaming startingOffsets latest不生效 旧Kafka消息延迟问题

1. startingOffsets 配置仅在流作业首次启动、对应checkpoint目录不存在时生效。只要你指定的HDFS checkpoint路径中存在历史运行产生的偏移量记录，Spark会优先从checkpoint存储的上次消费位置继续读取，完全忽略startingOffsets配置，这是你读到旧消息的核心原因。
2. 你的消费者配置存在参数拼写错误：正确的参数名为startingOffsets（复数形式），你配置的startingOffset（单数）属于无效配置，不会被Spark识别。
3. 你配置的consumer.override.auto.offset.reset是Kafka原生消费者参数，仅当Spark要读取的偏移量在Kafka集群中不存在时才会触发，正常场景下Spark自主管理偏移量时该参数不会生效。

• 先修正配置参数拼写：将消费者配置中的startingOffset改为startingOffsets，保证配置可以被正确识别。
• 如果你不需要任何历史消息，直接删除对应HDFS checkpoint路径下的所有文件，命令参考：hdfs dfs -rm -r <你的checkpointLocation路径>。删除后重启作业，此时无历史checkpoint记录，Spark会按照startingOffsets=latest的配置从最新消息开始消费。
• 如果你需要保留checkpoint中的其他状态数据不希望删除，可以在读取Kafka数据后增加时间过滤逻辑，直接丢弃早于30秒的消息，代码参考：

val readerB = spark
  .readStream
  .format("kafka")
  .options(consumerProperties)
  .load()
  .where($"timestamp" > current_timestamp() - expr("INTERVAL 30 SECONDS"))

该方案可以作为兜底逻辑，无论是否读到旧消息都会被过滤，完全匹配你的业务要求。

• 可选优化：增加maxOffsetsPerTrigger配置控制每批次拉取的最大消息量，避免一次性拉取过多历史消息导致作业延迟，保障实时性。

内容的提问来源于stack exchange，提问作者jcapi