[SPARK-51940][SS] Add interface for managing streaming checkpoint metadata

[jackierwzhang]

What changes were proposed in this pull request?

Minor refactor to introduce an interface for accessing the metadata (e.g. offset / commit logs) in a streaming checkpoint.

Why are the changes needed?

To standardize the access pattern.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

This is a pure refactoring, existing tests should suffice.

Was this patch authored or co-authored using generative AI tooling?

No.

[brkyvz]

LGTM! Thanks for the refactor. It makes access to the checkpoint outside of StreamExecution more formalized, e.g. in the State data store

[HeartSaVioR]

The code looks OK - meta question: I see you changed val to lazy val in multiple places. Is that intentional, and if it is, mind explaining?

[jackierwzhang]

The code looks OK - meta question: I see you changed val to lazy val in multiple places. Is that intentional, and if it is, mind explaining?

Yes, there are mainly two reason:

1. I don't want the initialization of the class to automatically initialize the logs since they involve FS operations, only direct access should initialize the logs.
2. Overriding a lazy val in StreamExecution is easier to reason that a plain val, which may be null when it's overridden by the subclass and hence throw NPE. For example, this id may throw null because checkpointMetadata above it is overridden.

[HeartSaVioR]

+1 pending CI. Thanks for the explanation.

[HeartSaVioR]

https://github.com/jackierwzhang/spark/actions/runs/14720115571/job/41312160109

Test failures seem to be relevant. @jackierwzhang Would you mind taking a look?

[jackierwzhang]

@HeartSaVioR I actually had to revert some lazy vals in StreamExecution back to vals because this access to id within streamMetadata would actually forces initialize the subclass's lazy val but at that time the input parameter asyncWritesExecutorService here is still null.

[HeartSaVioR]

@jackierwzhang Sorry, shall we retain the code comment as it is when you move this and below?

[jackierwzhang]

I removed it because they are duplicated with that in the class?

[HeartSaVioR]

Ah OK you moved the explanation to new class. Never mind.

[HeartSaVioR]

Thanks! Merging to master.