Why does the File Sink operation of Spark 2.4 Structured Streaming include double-level version validation?

Dataset<Row> dataset = this.sparkSession.readStream().format("kafka")
 .options(this.getSparkKafkaCommonOptions(sparkSession)) 
 .option("kafka.bootstrap.servers", "192.168.1.1:9092,192.168.1.2:9092")
 .option("subscribe", "myTopic1,myTopic2")
 .option("startingOffsets", "earliest")
 .load();

String mdtTempView = "mybasetemp";
 ExpressionEncoder<Row> Rowencoder = this.getSchemaEncoder(new Schema.Parser().parse(baseschema.getValue())); 
 Dataset<Row> parseVal = dataset.select("value").as(Encoders.BINARY())
 .map(new MapFunction<Row>(){
 ....
 }, Rowencoder)
 .createOrReplaceGlobalTempView(mdtTempView);
 
 Dataset<Row> queryResult = this.sparkSession.sql("select 。。。 from global_temp." + mdtTempView + " where start_time<>\"\"");

 String savePath= "/user/dx/streaming/data/testapp"; 
 String checkpointLocation= "/user/dx/streaming/checkpoint/testapp";

 StreamingQuery query = queryResult.writeStream().format("parquet")
 .option("path", savePath)
 .option("checkpointLocation", checkpointLocation)
 .partitionBy("month", "day", "hour")
 .outputMode(OutputMode.Append())
 .trigger(Trigger.ProcessingTime(1, TimeUnit.MINUTES))
 .start();
try {
 query.awaitTermination();
 } catch (StreamingQueryException e) {
 e.printStackTrace();
 }

class FileStreamSink(
 sparkSession: SparkSession,
 path: String,
 fileFormat: FileFormat,
 partitionColumnNames: Seq[String],
 options: Map[String, String]) extends Sink with Logging {
 private val basePath = new Path(path)
 private val logPath = new Path(basePath, FileStreamSink.metadataDir)
 private val fileLog =
 new FileStreamSinkLog(FileStreamSinkLog.VERSION, sparkSession, logPath.toUri.toString)
 
 override def addBatch(batchId: Long, data: DataFrame): Unit = {
   if (batchId <= fileLog.getLatest().map(_._1).getOrElse(-1L)) {
     logInfo(s"Skipping already committed batch $batchId")
   } else {
     // save file to hdfs
   }
 }
 //...
}