Spark-kafka - каталог контрольных точек, дублирующий данные

Question

Ниже приведен сценарий, который я тестировал

1. Я создал 200 записей, и он создал 20 файлов по 10 записей в каждом.
2. (после ожидания в течение 2 минут) я снова создал 200 записейи сразу убил приложение вручную (используя yarn -kill).На этот раз он создал только 7 файлов.
3. После перезапуска приложения он создал 20 файлов, вновь созданных на шаге 2.

В приведенном выше случае данные в 7 файлах дублируются. Как можноЯ управляю этим поведением, используя проверку наведения?

Ниже мой код:

 try{
  val kafkaOutput = result.writeStream
  .outputMode("append")
  .format("orc")
  .option("path", "/warehouse/test_duplicate/download/data1")
  .option("checkpointLocation", checkpoint_loc)
  .option("maxRecordsPerFile", 10)
  .trigger(Trigger.ProcessingTime("5 seconds"))
  .start()
  .awaitTermination() 

   result.checkpoint()

}catch {
        case e: Exception => e.printStackTrace

    }
finally
{
  println("**********Finally Called***********")
  result.checkpoint()
}