[apache/flink] [Flink-7218] [JobManager] ExecutionVertex.getPreferredLocationsBasedOnInputs() will always return empty

[周思华 <su...@163.com>]

I really like the new PR template, so i use it instead of the default one on github~

You can view, comment on, or merge this pull request online at: #4369 

What is the purpose of the change

The ExecutionVertex.getPreferredLocationsBasedOnInputs will always return empty, cause sourceSlot always be null until ExectionVertex has been deployed via 'Execution.deployToSlot()'. So allocate resource base on preferred location can't work correctly, we need to set the slot info for Execution as soon as Execution.allocateSlotForExecution() called successfully.

Brief change log
Added a field assignedFutureSlot in Execution to record the Future<SimpleSlot> as soon asExecution.allocateSlotForExecution() called successfully. And the assignedFutureSlot will be used in ExectionVertex. getPreferredLocationsBasedOnInputs () to get ExecutionVertex's preferred locations.
Verifying this change

This change added tests and can be verified as follows:

The test case is under ExecutionGraphSchedulingTest. testExecutionVertexGetPreferredLocationsBasedOnInputs(), i have simulated the process of the JobGraph deployment and validated the results in this test case.
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: (don't know)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes):
Documentation
Does this pull request introduce a new feature? (no)