[FLINK-19552][coordination] Consider only available input location preferences for slot profile in pipelined region scheduling

[azagrebin]

What is the purpose of the change

The pipelined region scheduling strategy schedules regions once all their input blocking dependencies are ready.
The SlotSharingGroups of the region can include executions of other regions which are not scheduled yet including their dependencies. Hence we should not wait other unavailable input dependencies to unblock the current region scheduling.

Brief change log

The PR moves PreferredLocationsRetriever creation from SchedulerBase to the ExecutionSlotAllocatorFactories.
Then the legacy allocator continues to use the DefaultPreferredLocationsRetriever.
The new SlotSharingExecutionSlotAllocator creates the DefaultSyncPreferredLocationsRetriever
where the original InputsLocationsRetriever is wrapped with the AvailableInputsLocationsRetriever.
It makes the InputsLocationsRetriever return only completed input location futures, others are filtered out.
This allows to return completed future from DefaultSyncPreferredLocationsRetriever and make it synchronous and non-blocking.

Verifying this change

unit tests

[flinkbot]

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit d4a0257 (Wed Oct 21 14:20:09 UTC 2020)

Warnings:

• No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

• ❓ 1. The [description] looks good.
• ❓ 2. There is [consensus] that the contribution should go into to Flink.
• ❓ 3. Needs [attention] from.
• ❓ 4. The change fits into the overall [architecture].
• ❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
• @flinkbot approve all to approve all aspects
• @flinkbot approve-until architecture to approve everything until architecture
• @flinkbot attention @username1 [@username2 ..] to require somebody's attention
• @flinkbot disapprove architecture to remove an approval you gave earlier

[azagrebin]

I also working on an integration test.
PipelinedRegionSchedulingConcurrentFailureTest, suggested in #13628, reproduces the exact original problem which is a combination of several discovered problems but the overall test looks a bit complicated.

I will try the following graph of disjoint pipelines:
v1 (ssg1) -> v2 (ssg3)
v3 (ssg2) -> v4 (ssg3)
with 2 slots.
e.g. v1 need to be canceled with its location and (v3 -> v2) region needs to be scheduled afterwards
then while deploying v2 and its SSG3, we will ask for v4 input (canceled v3).

[flinkbot]

CI report:

• 2f7b394 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[tillrohrmann]

Thanks for creating this PR @azagrebin. The changes look good to me. +1 for merging.

[zhuzhurk]

Thanks for opening this PR @azagrebin
I have a few minor comments for it.
The one I want to confirm is whether we can throw an exception when a bulk cannot be correctly created instead of just skipping scheduling the slot request check in e98f87a.

[rmetzger]

I have tested the change again with the failing benchmark job, and the JobManager doesn't exit with a FATAL error anymore.
No unexpected exceptions.

Thanks for the fix!

[rmetzger]

The behavior is still fine with the hotfix removed.

[azagrebin]

@flinkbot run azure