As I understand, you want to ingest "only new records coming through Kafka via old checkpoint path".

I would suggest that you can change .option("startingOffsets", "earliest") to .option("startingOffsets", "latest"). It would start reading from the newest offset and also avoid reprocessing of your old data. However, you would need to delete the old checkpoint path and start your stream after making the change.

P.S. - Don't try the above directly for production load.