[BP-1.19][FLINK-35522][runtime] Fix the issue that the source task may get stuck in speculative execution mode.

[SinBex]

What is the purpose of the change

If the source task does not get assigned a split because the SplitEnumerator has no more splits, and a failover occurs during the closing process, the SourceCoordinatorContext will not resend the NoMoreSplit event to the newly started source task, causing the source vertex to remain stuck indefinitely.
This case may only occur in batch jobs where speculative execution has been enabled.

Brief change log

• fix the issue that the source task may get stuck in speculative execution mode.

Verifying this change

• Added it case to verify the issue.

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:

• 310b4f0 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[zhuzhurk]

@flinkbot run azure

[zhuzhurk]

@flinkbot run azure

[SinBex]

@flinkbot run azure

[zhuzhurk]

LGTM.