[SPARK-XXXXX][SS] Use latestCommittedBatchId as currentBatchId when resuming late batch

[viirya]

What changes were proposed in this pull request?

This patch changes currentBatchId when MicroBatchExecution tries to resume from late batch from offset log. Previously it takes latestBatchId from offset log. This patch changes it to latestCommittedBatchId.

Why are the changes needed?

We have customer streaming job which is unable to restart from failed status after it failed to commit delta files. For example, if previous run failed to commit 14.delta, when the job restarted, it tried to read 14.delta. Because 14.delta doesn't exist (not committed), the job cannot be restarted to resume from late batch.

When MicroBatchExecution populates start offsets, it reads late batch id (latestBatchId) from offset log and committed batch id (latestCommittedBatchId) from commit log. Currently if latestCommittedBatchId == latestBatchId - 1, it means that we resume from late batch. But it uses latestBatchId as currentBatchId to run batch. Obviously, latestBatchId is 14 for above example, and latestCommittedBatchId is 13.

Because IncrementalExecution uses currentBatchId to load checkpointed states, it tries to load version 14 of delta files. But version 14 is not committed in late run. So resume run always fails because it cannot load non existing delta file.

We should use latestCommittedBatchId as currentBatchId instead of latestBatchId for the case resuming last batch.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Existing tests.

[viirya]

retest this please