[FLINK-9455][RM] Add support for multi task slot TaskExecutors #6734

tillrohrmann · 2018-09-22T17:17:56Z

What is the purpose of the change

This PR adds support for multi task slot TaskExecutors to Flink. Before it was recommended to start a Flink cluster with single slot TaskExecutors. Now also multi slot TaskExecutors can be configured and Flink won't allocate resources over-excessively.

Brief change log

Extend ResourceActions#allocateResource to return Collection<ResourceProfile> indicating the set of slots to expect
Store expected slots as PendingTaskManagerSlot
Use PendingTaskManagerSlot to fulfill PendingSlotRequest
Only ask for new resources if there are no more TaskManagerSlots and PendingTaskManagerSlots

Verifying this change

Added SlotManagerTest#: testRequestNewResources, testFailingAllocationReturnsPendingTaskManagerSlot, testPendingTaskManagerSlotCompletion, testRegistrationOfDifferentSlot, testOnlyFreeSlotsCanFulfillPendingTaskManagerSlot,

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, Yarn/Mesos, ZooKeeper: (yes)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (not applicable)

azagrebin

Thanks for the nice feature @tillrohrmann ! I left some comments to consider minor improvements before merge.

azagrebin · 2018-09-24T20:00:26Z

...esos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java

 		this.workersInLaunch = new HashMap<>(8);
 		this.workersBeingReturned = new HashMap<>(8);
+
+		final ContaineredTaskManagerParameters containeredTaskManagerParameters = taskManagerParameters.containeredParameters();


I would create a method to create and return slotsPerWorker to unload constructor code.

azagrebin · 2018-09-24T20:12:55Z

...esos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java

 		// create the specific TM parameters from the resource profile and some defaults
 		MesosTaskManagerParameters params = new MesosTaskManagerParameters(
-			resourceProfile.getCpuCores() < 1.0 ? taskManagerParameters.cpus() : resourceProfile.getCpuCores(),
+			taskManagerParameters.cpus(),


This looks like a copying of immutable MesosTaskManagerParameters. Is it on purpose?

No it is not on purpose. Will correct it.

azagrebin · 2018-09-25T08:14:48Z

...e/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/PendingSlotBalance.java

+		this(0, 0);
+	}
+
+	PendingSlotBalance(int numPendingSlots, int numUnassignedSlots) {


minor: @nonnegative

I will remove this class since it is a left-over from a previous implementation.

azagrebin · 2018-09-25T08:21:25Z

...e/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/PendingSlotRequest.java

+		return pendingTaskManagerSlot;
+	}
+
+	public void assignPendingTaskManagerSlot(PendingTaskManagerSlot pendingTaskManagerSlotToAssign) {


@nonnull pendingTaskManagerSlotToAssign seems to make sense only then

Yes will add it.

azagrebin · 2018-09-25T08:22:26Z

...c/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/PendingTaskManagerSlot.java

+		return resourceProfile;
+	}
+
+	public void assignPendingSlotRequest(PendingSlotRequest pendingSlotRequestToAssign) {


also here @nonnull pendingSlotRequestToAssign

azagrebin · 2018-09-25T08:41:45Z

...-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManager.java

+	}

-		updateSlot(slotId, allocationId, jobId);
+	private void completePendingTaskManagerSlot(ResourceProfile resourceProfile) {


is this completePendingTaskManagerSlot some leftover? looks unused

Yes, will remove it.

azagrebin · 2018-09-25T08:52:34Z

...n/java/org/apache/flink/runtime/resourcemanager/slotmanager/ImmutablePendingSlotBalance.java

+/**
+ * Immutable slot balance.
+ */
+final class ImmutablePendingSlotBalance extends PendingSlotBalance {


I cannot find where these balance classes are used. Are they for future?

No a left-over from a previous implementation. Will remove it.

azagrebin · 2018-09-25T10:01:32Z

...time/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerTest.java

+			slotManager.registerTaskManager(taskExecutorConnection, slotReport);
+
+			assertThat(slotManager.getNumberRegisteredSlots(), is(numberSlots - 1));
+			assertThat(slotManager.getNumberPendingTaskManagerSlots(), is(1));


The slot request should be fulfilled now, right? Then we could also check:
assertThat(slotManager.getNumberAssignedPendingTaskManagerSlots(), is(0));

I think getNumberPendingTaskManagerSlots == 1 is an even stricter condition, because it requires that actually some of the pending task manager slots have been completed.

azagrebin · 2018-09-25T10:10:37Z

...time/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerTest.java

-		final ResourceActions resourceManagerActions = mock(ResourceActions.class);
+		final AtomicInteger allocateResourceCalls = new AtomicInteger(0);
+		final ResourceActions resourceManagerActions = new TestingResourceActionsBuilder()
+			.setAllocateResourceConsumer(resourceProfile -> allocateResourceCalls.incrementAndGet())


the logic of some of this built TestingResourceActions repeat in some places, we could create some standard factory methods for them (maybe in future), e.g.:

resourceManagerActions = TestingResourceActionsBuilder.createAllocatingFixedSlotsResource(numOfSlotsPerWorker); allocateResourceCalls = resourceManagerActions.getAllocationsCounter();

True, one could add a helper function doing this.

GJL · 2018-09-26T18:28:06Z

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java

+	// ------------------------------------------------------------------------
+
+	protected static Collection<ResourceProfile> createSlotsPerWorker(int numSlots) {
+		final Collection<ResourceProfile> slots = new ArrayList<>(numSlots);


It would be enough to write:
Collections.nCopies(numSlots, ResourceProfile.ANY);

The returned List would also be immutable. At the moment there is no need to return a mutable list. I would even say it is not right that after calling Collection<ResourceProfile> startNewWorker(ResourceProfile resourceProfile), one may potentially modify the state of the ResourceManager (by accidentally modifying the returned collection).

Good point. I'll update the PR.

shuai-xu · 2018-09-27T08:17:11Z

...-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManager.java

+		final PendingTaskManagerSlot pendingTaskManagerSlot;
+
+		if (allocationId == null) {
+			pendingTaskManagerSlot = findExactlyMatchingPendingTaskManagerSlot(resourceProfile);


Why use exactly matching here but isMatching in other place? I think we should make the methods for finding pending task manager slots extendable so user can define their own matching algorithm according to need.

Because we use the ResourceProfile at the moment as a key for fulfilling pending task manager slots. Therefore, we need to find an exactly matching resource profile. In the future we could introduce an ID to identify newly started slots.

Think for example that you have two pending slots with resource profile (1,1) and (2,2). Now a slot with (2,2) is registered. You don't want to complete the pending task manager slot (1,1) (even though it would fulfill the requirements), because the slot (1,1), won't be able to complete the remaining pending task manager slot.

Making the matching algorithm pluggable should be a separate issue.

Extend ResourceActions interface to return a set of ResourceProfiles describing the set of slots with which the new resource will be started. The SlotManager stores them as PendingTaskManagerSlots which can be assigned to PendingSlotRequests. Only if there are no more TaskManagerSlots and PendingTaskManagerSlots, the SlotManager will request new resources from the ResourceManager. This closes apache#6734.

…State

…uration

tillrohrmann · 2018-09-27T15:28:37Z

Thanks for the review @azagrebin, @GJL and @shuai-xu. Merging this PR once Travis gives green light.

Extend ResourceActions interface to return a set of ResourceProfiles describing the set of slots with which the new resource will be started. The SlotManager stores them as PendingTaskManagerSlots which can be assigned to PendingSlotRequests. Only if there are no more TaskManagerSlots and PendingTaskManagerSlots, the SlotManager will request new resources from the ResourceManager. This closes apache#6734. (cherry picked from commit 771277b)

Myracle · 2020-01-06T09:11:44Z

@tillrohrmann Hello, I am confused with you code. Before your code, YarnResourceManager's requestYarnContainer will addContainerRequest only if pendingSlotRequests is bigger than pendingSlotAllocation, I think that Flink already won't allocate resources over-excessive when numberOfTaskSlots is bigger than 1. So I am confused what your code does compared with the previous code. Thank you very much.

tillrohrmann · 2020-01-06T10:42:48Z

Hi @Myracle, the problem was that Flink did not take the number of pending slots into account (one TaskExecutor can be started with many slots) and instead only looked at the number of container requests (== number of TaskExecutors to be started). With this change, Flink does this properly.

Myracle · 2020-01-06T11:11:43Z

@tillrohrmann Thank you for your reply. Before your change, Flink calculate pendingTaskManagerSlots(==numPendingContainerRequests * numberOfTaskSlots) in the class YarnResourceManager. Flink will only request containers from yarn when pendingTaskManagerSlots is bigger than pendingTaskManagerSlots. It means that Flink will request resources only when total requested slots can not satisfy current's requirement. There is another issue ([FLINK-9567][yarn] Before requesting new containers always check if it is required). Does FLINK-9567 resolve the same problem? What's the difference between your issue and FLINK-9567? Thank you.

azagrebin approved these changes Sep 25, 2018

View reviewed changes

tillrohrmann force-pushed the multiSlotSupport branch from a59e4d4 to c5412f5 Compare September 26, 2018 15:35

GJL requested changes Sep 26, 2018

View reviewed changes

shuai-xu reviewed Sep 27, 2018

View reviewed changes

tillrohrmann force-pushed the multiSlotSupport branch from c5412f5 to 3201620 Compare September 27, 2018 10:08

GJL approved these changes Sep 27, 2018

View reviewed changes

tillrohrmann force-pushed the multiSlotSupport branch from 3201620 to 2a7c6b9 Compare September 27, 2018 15:27

tillrohrmann added 5 commits September 27, 2018 17:27

[hotfix] Cancel actual pending slot request in SlotManager#updateSlot…

43a96bb

…State

[hotfix] Remove mocking from SlotManagerTest

844a0b6

[hotfix] Remove mocking from SlotProtocolTest

6ba4563

[hotfix] Start MesosWorkers with default ContaineredTaskManagerConfig…

1ac7336

…uration

tillrohrmann force-pushed the multiSlotSupport branch from 2a7c6b9 to 1ac7336 Compare September 27, 2018 15:28

asfgit closed this in 771277b Sep 28, 2018

tillrohrmann deleted the multiSlotSupport branch September 28, 2018 09:40

rmetzger added the component=Runtime/Coordination label Mar 18, 2019

[FLINK-9455][RM] Add support for multi task slot TaskExecutors #6734

[FLINK-9455][RM] Add support for multi task slot TaskExecutors #6734

Uh oh!

Conversation

tillrohrmann commented Sep 22, 2018

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

azagrebin left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

GJL Sep 26, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tillrohrmann commented Sep 27, 2018

Uh oh!

Myracle commented Jan 6, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

tillrohrmann commented Jan 6, 2020

Uh oh!

Myracle commented Jan 6, 2020

Uh oh!

Uh oh!

GJL Sep 26, 2018 •

edited

Loading

Myracle commented Jan 6, 2020 •

edited

Loading