[FLINK-38497][connector] FLIP-535:Introduce RateLimiter to Source.

[lvyanquan]

What is the purpose of the change

Provide a configuration solution for fine-grained read speed control in the Flink Source connector.

Brief change log

• Add new constructor for SourceReaderBase and SingleThreadMultiplexSourceReaderBase
• Add CompletionStage acquire(int requestSize) and void notifyStatusChange(S status) method in RateLimiter

Verifying this change

This change added tests and can be verified as follows:

• Added test in SourceReaderBaseTest that validates that Records were emitted with rate control

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): (yes)
• The serializers: (no)
• The runtime per-record code paths (performance sensitive): (yes)
• 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)
• If yes, how is the feature documented? (JavaDocs)

[flinkbot]

CI report:

• 68970ea UNKNOWN
• 62850fa Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[lvyanquan]

Hi @leonardBang @ruanhang1993, maybe you can help to review this.

[leonardBang]

Thanks @lvyanquan for the contribution, I left some comments. And you need to open a task to add documentation for new feature

[davidradl]

When we say requestSize - does this mean the size of requests? Looking at the code it seems to relate to the number of futures we want to acquire, if so I would suggest renaming this. Maybe numberOfFuturesToAcquire.

[lvyanquan]

The returned Future here represents a single asynchronous execution result, and there is no scenario where multiple Future instances are returned. Therefore, the variable name numberOfFutures is not appropriate.

The input parameter here represents the number of requests, which could specifically refer to the number (or even size) of data records. Since this interface is not exclusively called by SourceReader, I used the more generic term "Request" to abstract the concept beyond data-specific contexts.

[davidradl]

Ok thanks for the clarification. numberOfRequests seems better to me as it is not todo the with the size of individual requests. It is up to you if you want to change or not.

[leonardBang]

+1 to use numberOfRequests @lvyanquan

[lvyanquan]

Agree to use numberOf instead of size.
Note that the original comment for the acquire() method states:

flink/flink-core/src/main/java/org/apache/flink/api/connector/source/util/ratelimit/RateLimiter.java

Line 33 in 7b0a882

* Returns a future that is completed once another event would not exceed the rate limit. For

Using event is more appropriate than request in this context, hence the term numberOfEvents is used now.

[lvyanquan]

Thanks for suggestion of @davidradl and @leonardBang, updated.

[lvyanquan]

@flinkbot run azure

[lvyanquan]

Subtask for document: https://issues.apache.org/jira/browse/FLINK-38589.

[lvyanquan]

@flinkbot run azure

[lvyanquan]

Rebase master to fix CI failure of StreamExecutionEnvironmentTests::test_set_requirements_with_cached_directory.

[lvyanquan]

Updated based on comments.

[lvyanquan]

@flinkbot run azure

[leonardBang]

Thanks @lvyanquan for the update, generally looks good to me, merging...