[FLINK-40234] Add unit test to demonstrate that busy subtask thread may cause splits to be marked as idle erroneously and cause dropped records

[leekeiabstraction]

What is the purpose of the change

Add unit test to demonstrate that busy subtask thread may cause splits to be marked as idle erroneously and cause dropped records. See https://issues.apache.org/jira/browse/FLINK-40234

Brief change log

Add the testMultiSplitWithBusySubtask parameterised test case

• The test case passes when subtask busy-ness is NOT simulated
• The test case fails (split marked as idle and records were emitted behind watermark) when the subtask busyness is simulated. This demonstrates that subtask busy-ness can impact the semantic of a Flink job

Verifying this change

This PR adds test cases to demonstrate a bug.

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: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Sonnet 5

[flinkbot]

CI report:

• 1b1f39a Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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