[FLINK-38564][connector] FLIP-537: Enumerator Maintains Global Splits Distribution for Balanced assignment.

[loserwang1024]

What is the purpose of the change

As shown in https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=373886480, add a new source contract that enables the enumerator to have a complete global view of splits.

Brief change log

Extending FLIP-27， this FLIP modified ReaderRegistrationEvent Contract:

If a Source implements SupportsSplitReassignmentOnRecovery, the SourceOperator does not assigned splits from state to the reader. Instead, it reports the splits via a ReaderRegistrationEvent sent to the SplitEnumerator.If a Source is not implements SupportsSplitReassignmentOnRecovery, the SourceOperator assigned splits from state to the reader.The splits won't be included in ReaderRegistrationEvent.

After the SourceCoordinator receives the ReaderRegistrationEvent, it calls addReader on the SplitEnumerator. At this point, the enumerator can access the reported splits through the ReaderInfo obtained from SplitEnumeratorContext#registeredReaders.

Verifying this change

• SourceCoordinatorTest#testDuplicateRedistribution
• SourceOperatorTest#testOpen

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): 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
• If yes, how is the feature documented? JavaDocs

[flinkbot]

CI report:

• abe1bc7 Azure: FAILURE
• 0f9b348 Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

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

[loserwang1024]

@leonardBang @ruanhang1993 ,CC

[loserwang1024]

@flinkbot run azure

[leonardBang]

Thanks @loserwang1024 for the contribution, I left some comments

[leonardBang]

reassignSplits(unassignedSplits) ?

[loserwang1024]

Maybe recalculateAssignments? Don't assign them until the split is added.

[loserwang1024]

@leonardBang Thanks for your review. Just modify it.