[FLINK-19151][yarn]Fix the unit value according to different yarn scheduler #13347

jinhai-cloud · 2020-09-08T02:55:58Z

What is the purpose of the change

*Currently, Flink matches the allocated containers to the original requests by reading the Yarn configurations and calculate how the requested resources should be normalized.

What has been overlooked is that, Yarn FairScheduler (and its subclass SLSFairScheduler) has overridden the normalization behavior.

Flink should be good enough to cover behaviors of all the schedulers that Yarn currently provides*

Brief change log

add method to get unit resource in YarnResourceManager
add two fields to WorkerSpecContainerResourceAdapter: unitMemMB and unitVcore

Verifying this change

This change is already covered by existing tests.

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

Documentation

Does this pull request introduce a new feature? (no)

flinkbot · 2020-09-08T02:57:45Z

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 08cde36 (Tue Sep 08 02:57:45 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.

Details

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

flinkbot · 2020-09-08T03:16:48Z

CI report:

69e1ce2 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

xintongsong

Thanks for preparing this PR, @Manbuyun.
I have some comments, which would be good to address before merging this.
In addition to the inline comment:

There's no testing cases that verify the changes
The commit messages are not following the community's conventions. Please refer to "Commit Naming Conventions" in this page

flink-yarn/src/main/java/org/apache/flink/yarn/WorkerSpecContainerResourceAdapter.java

flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java

xintongsong

Thanks for addressing my comments, @Manbuyun.
The PR already looks quite good to me now. I'll take over from here.
I have only a few minor comments, which I will address myself while merging this.

xintongsong · 2020-09-10T02:56:35Z

flink-yarn/src/main/java/org/apache/flink/yarn/WorkerSpecContainerResourceAdapter.java

 	}

 	/**
 	 * Normalize to the minimum integer that is greater or equal to 'value' and is positive integer multiple of 'unitValue'.


JavaDoc needs update.

xintongsong · 2020-09-10T03:18:31Z

flink-yarn/src/test/java/org/apache/flink/yarn/UtilsTest.java

+	public void testGetUnitResource() {
+		YarnConfiguration yarnConfig = new YarnConfiguration();
+
+		yarnConfig.set("yarn.resourcemanager.scheduler.class", "org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler");


We should always try to avoid magic string literals if possible.

xintongsong · 2020-09-10T04:03:21Z

flink-yarn/src/test/java/org/apache/flink/yarn/WorkerSpecContainerResourceAdapterTest.java

+		final int unitMemMB1 = 150;
+		final int unitVcore1 = 15;
+		final WorkerSpecContainerResourceAdapter adapter1 =
+			new WorkerSpecContainerResourceAdapter(
+				getConfigProcessSpecEqualsWorkerSpec(),
+				minMemMB,
+				minVcore,
+				Integer.MAX_VALUE,
+				Integer.MAX_VALUE,
+				unitMemMB1,
+				unitVcore1,
+				Collections.emptyMap());
+
+		final Resource containerResource4 = Resource.newInstance(150, 15);
+
+		assertThat(adapter1.getWorkerSpecs(containerResource1, strategy), empty());
+		assertThat(adapter1.getWorkerSpecs(containerResource2, strategy), empty());
+
+		assertThat(adapter1.tryComputeContainerResource(workerSpec1).get(), is(containerResource4));
+		assertThat(adapter1.tryComputeContainerResource(workerSpec2).get(), is(containerResource4));
+		assertThat(adapter1.tryComputeContainerResource(workerSpec3).get(), is(containerResource4));
+		assertThat(adapter1.tryComputeContainerResource(workerSpec4).get(), is(containerResource4));
+
+		assertThat(adapter1.getWorkerSpecs(containerResource4, strategy), containsInAnyOrder(workerSpec1, workerSpec2, workerSpec3, workerSpec4));


I'm not sure whether it is necessary to introduce another adapter. Should be good enough to tune the parameters of existing cases.

I guess the problem is that the original test case did not make its intention explicit.

xintongsong · 2020-09-10T04:03:42Z

flink-yarn/src/test/java/org/apache/flink/yarn/WorkerSpecContainerResourceAdapterTest.java

+		final int unitMemMB1 = 150;
+		final int unitVcore1 = 15;
+		final WorkerSpecContainerResourceAdapter adapter1 =
+			new WorkerSpecContainerResourceAdapter(
+				getConfigProcessSpecEqualsWorkerSpec(),
+				minMemMB,
+				minVcore,
+				Integer.MAX_VALUE,
+				Integer.MAX_VALUE,
+				unitMemMB1,
+				unitVcore1,
+				Collections.emptyMap());
+
+		final Resource containerResource6 = Resource.newInstance(150, 15);
+
+		assertThat(adapter1.tryComputeContainerResource(workerSpec1).get(), is(containerResource6));
+		assertThat(adapter1.tryComputeContainerResource(workerSpec2).get(), is(containerResource6));
+		assertThat(adapter1.tryComputeContainerResource(workerSpec3).get(), is(containerResource6));
+		assertThat(adapter1.tryComputeContainerResource(workerSpec4).get(), is(containerResource6));
+
+		assertThat(adapter1.getEquivalentContainerResource(containerResource4, strategy), empty());
+		assertThat(adapter1.getEquivalentContainerResource(containerResource5, strategy), contains(containerResource6));
+
+		assertThat(adapter1.getWorkerSpecs(containerResource3, strategy), containsInAnyOrder(workerSpec1, workerSpec2, workerSpec3, workerSpec4));


…, with respect to Yarn FairScheduler.

…, with respect to Yarn FairScheduler. This closes #13347.

rmetzger added the review=description? label Sep 8, 2020

rmetzger added the component=Deployment/YARN label Sep 8, 2020

xintongsong requested changes Sep 8, 2020

View reviewed changes

xintongsong approved these changes Sep 10, 2020

View reviewed changes

[FLINK-19151][yarn] Update container resource normalization algorithm…

69e1ce2

…, with respect to Yarn FairScheduler.

xintongsong closed this in 00bf41f Sep 10, 2020

xintongsong pushed a commit that referenced this pull request Sep 10, 2020

[FLINK-19151][yarn] Update container resource normalization algorithm…

bfff6b1

…, with respect to Yarn FairScheduler. This closes #13347.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[FLINK-19151][yarn]Fix the unit value according to different yarn scheduler #13347

[FLINK-19151][yarn]Fix the unit value according to different yarn scheduler #13347

Uh oh!

jinhai-cloud commented Sep 8, 2020

Uh oh!

flinkbot commented Sep 8, 2020

Uh oh!

flinkbot commented Sep 8, 2020 •

edited

Loading

Uh oh!

xintongsong left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

xintongsong left a comment

Uh oh!

xintongsong Sep 10, 2020

Uh oh!

xintongsong Sep 10, 2020

Uh oh!

xintongsong Sep 10, 2020

Uh oh!

xintongsong Sep 10, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

[FLINK-19151][yarn]Fix the unit value according to different yarn scheduler #13347

[FLINK-19151][yarn]Fix the unit value according to different yarn scheduler #13347

Uh oh!

Conversation

jinhai-cloud commented Sep 8, 2020

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!

flinkbot commented Sep 8, 2020

Automated Checks

Review Progress

Uh oh!

flinkbot commented Sep 8, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

xintongsong left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

xintongsong left a comment

Choose a reason for hiding this comment

Uh oh!

xintongsong Sep 10, 2020

Choose a reason for hiding this comment

Uh oh!

xintongsong Sep 10, 2020

Choose a reason for hiding this comment

Uh oh!

xintongsong Sep 10, 2020

Choose a reason for hiding this comment

Uh oh!

xintongsong Sep 10, 2020

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

flinkbot commented Sep 8, 2020 •

edited

Loading