I found the problem why the code didn't trigger the bad records write. I did empty the folder for bad records. After fixing that, it works. Thanks for the help Isi 🙂

    data_frame.write.format("delta").option("optimizeWrite", "true").mode(
        "overwrite"
    ).saveAsTable(table_name)