[FLINK-29869][ResourceManager] make ResourceAllocator declarative.

[huwh]

What is the purpose of the change

This pull request makes ResourceAllocator declarative in order to reduce the coupling between the SlotManager and the ResourceManager

Brief change log

• Introduce declareResourceNeeded into ResourceAllocator. ActiveResourceManager will allocate/release workers to meet the declared resources.
• SlotManager use declareResourceNeeded instead of allocate/releaseResource to reduce the coupling between the SlotManager and the ResourceManager

Verifying this change

This change added tests and can be verified as follows:

• Added tests that validates that ResourceManager will allocate/release workers to meet the declared resources.
• Update the tests for SlotManager to use the declarative api.

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: (yes, ResourceManager/SlotManager)
• The S3 file system connector: (no)

Documentation

• Does this pull request introduce a new feature? (yes)
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
  • https://docs.google.com/document/d/1lcmf3MKmcmf9tsPc1whaZHMYKurGoGtqOXEep9ngP2k/edit

[flinkbot]

CI report:

• bb85fd1 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[huwh]

@xintongsong Could you please help review this PR in your free time?

[huwh]

@flinkbot run azure

[xintongsong]

Thanks for opening this PR, @huwh. I have some questions. Please take a look.

[huwh]

Thanks for your comment, @xintongsong, I made the changes

1. delete UnwantedWorker and UnwantedWorkerWithResourceProfile. Make the unwanted workers as hint.
2. add totalWorkerCounter to maintian the count of all workers which have resource spec(include recovered workers)
3. remove the delayed check of declarationResource, since its cost is not very heavy.

[huwh]

@xintongsong thanks for the review. problems are addressed. In addition, ARM#requestWorkerIfRequired has been replaced by checkResourceDeclarations at this PR.

[huwh]

I have manually verified these tests below, with DeclarativeSlotManager and DeclarativeSlotManager

• job with 2 parallism, flink cluster with 2 TaskManagers, each have 1 slot.
• TaskManager could released when idle timeoued.
• TaskManager could released when registered slots more than max.
• New JobManager/RM will use the previous TM when JobManager failover.
• New TaskManager will be requested when TaskManager terminated.
  • registered taskmanager terminated
  • allocated not registered taskmanager terminated
  • released by RM because of reaching resourcemanager.taskmanager-registration.timeout

[xintongsong]

@huwh, thanks for addressing the comments. I have squashed the previous fix-up commits to get a complete view of the changes. I think this PR is very close to a mergable state. I have only one concern regarding the semantic change of SlotManager#registerTaskManager, plus a few minor comments.

[xintongsong]

The xxxWithDelay pattern has been repeated 4 times (declarative/fine-grained slot manager, check resource requirement / declare needed resources), and there could be more in future. I wonder if it can be deduplicated as a util. Not necessary for this PR though. This PR is already quite complicated.

[huwh]

Thanks, @xintongsong, comments are addressed except the XXXWithDely, we can address this in another PR.

I force pushed the new commits since it is not easy to avoid conflict while move TestingResourceAllocator.java#declareResourceNeededConsumer to the previous commit.

[xintongsong]

Thanks @huwh. LGTM. Merging this once CI passes.