[FLINK-30561][state/changelog] fix changelog local cache file not found

[zoltar9264]

What is the purpose of the change

Fix the changelog local cache file not found while restarting, described in FLINK-30561.

Brief change log

(for example:)

• prevent StateChangelogHandleStreamHandleReader close current iterator twice.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
• The serializers: (no)
• The runtime per-record code paths (performance sensitive): (no)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
• The S3 file system connector: (no)

Documentation

• Does this pull request introduce a new feature? (no)

[flinkbot]

CI report:

• 7d6c047 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[masteryhx]

Thanks for the pr.
It's a bit strange that add close flag in StateChangelogHandleStreamHandleReader.
I'd like to suggest to add this flag in the stream of ChangelogStreamHandleReaderWithCache:

1. IIUC, Upon restart, Only ChangelogStreamHandleReaderWithCache will be global, but StateChangelogHandleStreamHandleReader will always be used as a new object so that current flag will not be reused and will always be false ?
2. It's better to put the responsbility of "close once" into the smallest scope who relys on it.

Or maybe I missed something ?

[zoltar9264]

Thanks for your suggestion @masteryhx . I agree with the second point. As for the first point, whether StateChangelogHandleStreamHandleReader is reused after restart will cause this problem.

I will move the close flag to ChangelogStreamHandleReaderWithCache later.

[rkhachatryan]

Thanks for the PR @zoltar9264 .

I agree with @masteryhx and seems like you've already addressed his comment.

I have another question about concurrency, PTAL.

Besides of that, would that be possible and not too hard to write a test for this case?

[rkhachatryan]

LGTM