[FLINK-38735][connector-file] add restore state check

[liunaijie]

What is the purpose of the change

bug fix for https://issues.apache.org/jira/browse/FLINK-38735

Brief change log

• Avoid got NoSuchElementException when restore with empty state

Verifying this change

WIP

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 : yes

[flinkbot]

CI report:

• 2a813b1 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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

[davidradl]

1. Please correct the title to include the jira number
2. I see you indicate htis is covered by existing tests, but the existing tests did not fail, can we add a test to restore with empty state to drive the failing case.

[liunaijie]

Thanks for your review and quick response. @davidradl

1. Please correct the title to include the jira number

Done

2. I see you indicate htis is covered by existing tests, but the existing tests did not fail, can we add a test to restore with empty state to drive the failing case.

Sure I will try to add a test case to cover this edge case.

[liunaijie]

I write a test with mockito, the result was as we expected.

Before fix, when restore with empty state, it will throw NoSuchElementException.

After fix, the class init will success, and can get empty result.

As flink does not recommend using mockito to write unit tests, so I need some time to learn how to write this test.