You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/09/16 06:27:17 UTC
[GitHub] [flink] xintongsong opened a new pull request #13397: [FLINK-19177][runtime] Extend managed memory weight/fraction interfaces for various use cases.
xintongsong opened a new pull request #13397:
URL: https://github.com/apache/flink/pull/13397
## What is the purpose of the change
This PR extends interfaces for getting/setting managed memory weights/fractions with respect to various use cases. It also introduces a configuration option setting use case weights.
This PR does not touch the fraction calculations (which is changed in a separate following PR), thus should not change any behaviors in production.
## Brief change log
- 6a178ff1a861550d36984fae1fdd4c39f1ed992e: Disable fraction calculation for fine grained resource specs, because the original calculation does not work with various use cases.
- b1a0d26eda1107f5475834c414f218afc8fbdc1f: Introduce `ManagedMemoryUseCase`
- 2f4011193d0020d461de81049499387c3bea0db6..96d78c11c6e6aaf408504d45d70a7e69283f016c: Extend managed memory weight/fraction interfaces in `Transformation`, `StreamNode`, `StreamConfig` to support various use cases.
- 1e0209f0df58ecdc3460d426262fc89d035b83f6: Introduce configuration option for setting use case weights
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on a change in pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
xintongsong commented on a change in pull request #13397:
URL: https://github.com/apache/flink/pull/13397#discussion_r489912221
##########
File path: flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
##########
@@ -401,6 +403,22 @@
.withDescription("Fraction of Total Flink Memory to be used as Managed Memory, if Managed Memory size is not"
+ " explicitly specified.");
+ /**
+ * Weights of managed memory consumers.
+ */
+ // Do not advertise this option until the feature is completed.
+ @Documentation.ExcludeFromDocumentation
+ public static final ConfigOption<Map<String, String>> MANAGED_MEMORY_CONSUMER_WEIGHTS =
+ key("taskmanager.memory.managed.consumer-weights")
+ .mapType()
+ .defaultValue(new HashMap<String, String>() {{
+ put("DATAPROC", "70");
+ put("PYTHON", "30");
Review comment:
Since map type `ConfigOption` only accepts `Map<String, String>`, I'll use a static inner class with static final string constants for the consumer names, to avoid converting enum values to strings.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13397: [FLINK-19177][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13397:
URL: https://github.com/apache/flink/pull/13397#issuecomment-693202484
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 1e0209f0df58ecdc3460d426262fc89d035b83f6 (Wed Sep 16 06:29:51 UTC 2020)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## 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](https://flink.apache.org/contributing/reviewing-prs.html) 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 <summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13397:
URL: https://github.com/apache/flink/pull/13397#issuecomment-693261990
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6550",
"triggerID" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"triggerType" : "PUSH"
}, {
"hash" : "b8e95c022986da9f0b0d1b18e205f7cbc05f51ab",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6578",
"triggerID" : "b8e95c022986da9f0b0d1b18e205f7cbc05f51ab",
"triggerType" : "PUSH"
}, {
"hash" : "7c2c20f97637a965be7c74243592f1157d1ac9f7",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6583",
"triggerID" : "7c2c20f97637a965be7c74243592f1157d1ac9f7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b8e95c022986da9f0b0d1b18e205f7cbc05f51ab Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6578)
* 7c2c20f97637a965be7c74243592f1157d1ac9f7 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6583)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on a change in pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
xintongsong commented on a change in pull request #13397:
URL: https://github.com/apache/flink/pull/13397#discussion_r489895561
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
##########
@@ -177,11 +178,13 @@ private JobGraph createJobGraph() {
setSlotSharingAndCoLocation();
+ // For now, only consider managed memory for batch algorithms.
+ // TODO: extend managed memory fraction calculations w.r.t. various managed memory use cases.
setManagedMemoryFraction(
Collections.unmodifiableMap(jobVertices),
Collections.unmodifiableMap(vertexConfigs),
Collections.unmodifiableMap(chainedConfigs),
- id -> streamGraph.getStreamNode(id).getManagedMemoryWeight());
+ id -> streamGraph.getStreamNode(id).getManagedMemoryUseCaseWeights().getOrDefault(ManagedMemoryUseCase.BATCH_OP, 0));
Review comment:
Not sure about this. I think we don't really want a default weight here.
The purpose for this default is to avoid NPE when performing `mapToInt` and `sum` in `setManagedMemoryFractionForSlotSharingGroup`.
Maybe we should not set default value here and handles the `null` values in `setManagedMemoryFractionForSlotSharingGroup`. I think it should be straightforward to convert `null` to `0` for a `sum` processing.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13397:
URL: https://github.com/apache/flink/pull/13397#issuecomment-693261990
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6550",
"triggerID" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"triggerType" : "PUSH"
}, {
"hash" : "b8e95c022986da9f0b0d1b18e205f7cbc05f51ab",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6578",
"triggerID" : "b8e95c022986da9f0b0d1b18e205f7cbc05f51ab",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1e0209f0df58ecdc3460d426262fc89d035b83f6 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6550)
* b8e95c022986da9f0b0d1b18e205f7cbc05f51ab Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6578)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on a change in pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
xintongsong commented on a change in pull request #13397:
URL: https://github.com/apache/flink/pull/13397#discussion_r489915944
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
##########
@@ -849,26 +843,23 @@ private static void setManagedMemoryFractionForSlotSharingGroup(
}
private static void setManagedMemoryFractionForOperator(
- final ResourceSpec operatorResourceSpec,
final ResourceSpec groupResourceSpec,
final int operatorManagedMemoryWeight,
final int groupManagedMemoryWeight,
final StreamConfig operatorConfig) {
- final double managedMemoryFraction;
-
if (groupResourceSpec.equals(ResourceSpec.UNKNOWN)) {
- managedMemoryFraction = groupManagedMemoryWeight > 0
- ? getFractionRoundedDown(operatorManagedMemoryWeight, groupManagedMemoryWeight)
- : 0.0;
+ operatorConfig.setManagedMemoryFraction(
+ groupManagedMemoryWeight > 0 ?
+ getFractionRoundedDown(operatorManagedMemoryWeight, groupManagedMemoryWeight) :
+ 0.0);
} else {
- final long groupManagedMemoryBytes = groupResourceSpec.getManagedMemory().getBytes();
- managedMemoryFraction = groupManagedMemoryBytes > 0
- ? getFractionRoundedDown(operatorResourceSpec.getManagedMemory().getBytes(), groupManagedMemoryBytes)
- : 0.0;
+ // Supporting for fine grained resource specs is still under developing.
+ // This branch should not be executed in production. Not throwing exception for testing purpose.
+ LOG.error("Failed setting managed memory fractions. " +
+ " Operators may not be able to use managed memory properly." +
+ " Calculating managed memory fractions with fine grained resource spec is currently not supported.");
Review comment:
FYI, I've created FLINK-19267 to track the fraction calculation issue for fine grained resource specs.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13397:
URL: https://github.com/apache/flink/pull/13397#issuecomment-693261990
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6550",
"triggerID" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1e0209f0df58ecdc3460d426262fc89d035b83f6 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6550)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13397:
URL: https://github.com/apache/flink/pull/13397#issuecomment-693261990
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6550",
"triggerID" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"triggerType" : "PUSH"
}, {
"hash" : "b8e95c022986da9f0b0d1b18e205f7cbc05f51ab",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6578",
"triggerID" : "b8e95c022986da9f0b0d1b18e205f7cbc05f51ab",
"triggerType" : "PUSH"
}, {
"hash" : "7c2c20f97637a965be7c74243592f1157d1ac9f7",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "7c2c20f97637a965be7c74243592f1157d1ac9f7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b8e95c022986da9f0b0d1b18e205f7cbc05f51ab Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6578)
* 7c2c20f97637a965be7c74243592f1157d1ac9f7 UNKNOWN
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on a change in pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
xintongsong commented on a change in pull request #13397:
URL: https://github.com/apache/flink/pull/13397#discussion_r489886220
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
##########
@@ -849,26 +843,23 @@ private static void setManagedMemoryFractionForSlotSharingGroup(
}
private static void setManagedMemoryFractionForOperator(
- final ResourceSpec operatorResourceSpec,
final ResourceSpec groupResourceSpec,
final int operatorManagedMemoryWeight,
final int groupManagedMemoryWeight,
final StreamConfig operatorConfig) {
- final double managedMemoryFraction;
-
if (groupResourceSpec.equals(ResourceSpec.UNKNOWN)) {
- managedMemoryFraction = groupManagedMemoryWeight > 0
- ? getFractionRoundedDown(operatorManagedMemoryWeight, groupManagedMemoryWeight)
- : 0.0;
+ operatorConfig.setManagedMemoryFraction(
+ groupManagedMemoryWeight > 0 ?
+ getFractionRoundedDown(operatorManagedMemoryWeight, groupManagedMemoryWeight) :
+ 0.0);
} else {
- final long groupManagedMemoryBytes = groupResourceSpec.getManagedMemory().getBytes();
- managedMemoryFraction = groupManagedMemoryBytes > 0
- ? getFractionRoundedDown(operatorResourceSpec.getManagedMemory().getBytes(), groupManagedMemoryBytes)
- : 0.0;
+ // Supporting for fine grained resource specs is still under developing.
+ // This branch should not be executed in production. Not throwing exception for testing purpose.
+ LOG.error("Failed setting managed memory fractions. " +
+ " Operators may not be able to use managed memory properly." +
+ " Calculating managed memory fractions with fine grained resource spec is currently not supported.");
Review comment:
Adding a `TODO` message and creating a follow-up issue sound good to me.
The problem of throwing an exception is that, existing test cases using fine grained resource specs will fail on generating job graphs.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] KarmaGYZ commented on a change in pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
KarmaGYZ commented on a change in pull request #13397:
URL: https://github.com/apache/flink/pull/13397#discussion_r489294355
##########
File path: flink-core/src/main/java/org/apache/flink/api/dag/Transformation.java
##########
@@ -139,12 +142,12 @@ public static int getNewNodeId() {
private ResourceSpec preferredResources = ResourceSpec.DEFAULT;
/**
- * This weight indicates how much this transformation relies on managed memory, so that
- * transformation highly relies on managed memory would be able to acquire more managed
- * memory in runtime (linear association). Note that it only works in cases of UNKNOWN
- * resources.
+ * Each entry in this map represents a use case that this transformation needs managed memory for. The key of the
+ * entry indicates the use case, while the value is the use-case-specific weight for this transformation. Managed
+ * memory reserved for an OPERATOR scope use case will be shared by all the declaring transformations within a slot
+ * according to this weight. For SLOT scope use cases, the weights are ignored.
*/
- private int managedMemoryWeight = DEFAULT_MANAGED_MEMORY_WEIGHT;
+ private final Map<ManagedMemoryUseCase, Integer> managedMemoryUseCaseWeights = new HashMap<>();
Review comment:
Just trying to understand, is it illegal to put `ROCKSDB` or `PYTHON` in this map in your design? If so, should we check the sanity if someone tries to do it?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
##########
@@ -849,26 +843,23 @@ private static void setManagedMemoryFractionForSlotSharingGroup(
}
private static void setManagedMemoryFractionForOperator(
- final ResourceSpec operatorResourceSpec,
final ResourceSpec groupResourceSpec,
final int operatorManagedMemoryWeight,
final int groupManagedMemoryWeight,
final StreamConfig operatorConfig) {
- final double managedMemoryFraction;
-
if (groupResourceSpec.equals(ResourceSpec.UNKNOWN)) {
- managedMemoryFraction = groupManagedMemoryWeight > 0
- ? getFractionRoundedDown(operatorManagedMemoryWeight, groupManagedMemoryWeight)
- : 0.0;
+ operatorConfig.setManagedMemoryFraction(
+ groupManagedMemoryWeight > 0 ?
+ getFractionRoundedDown(operatorManagedMemoryWeight, groupManagedMemoryWeight) :
+ 0.0);
} else {
- final long groupManagedMemoryBytes = groupResourceSpec.getManagedMemory().getBytes();
- managedMemoryFraction = groupManagedMemoryBytes > 0
- ? getFractionRoundedDown(operatorResourceSpec.getManagedMemory().getBytes(), groupManagedMemoryBytes)
- : 0.0;
+ // Supporting for fine grained resource specs is still under developing.
+ // This branch should not be executed in production. Not throwing exception for testing purpose.
+ LOG.error("Failed setting managed memory fractions. " +
+ " Operators may not be able to use managed memory properly." +
+ " Calculating managed memory fractions with fine grained resource spec is currently not supported.");
Review comment:
I agree that the solution for supporting fine-grained resource specs should not block this work. However, we might create a ticket to track that work and mention it in the comment. We could also add the `TODO` label to the comment.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
##########
@@ -849,26 +843,23 @@ private static void setManagedMemoryFractionForSlotSharingGroup(
}
private static void setManagedMemoryFractionForOperator(
- final ResourceSpec operatorResourceSpec,
final ResourceSpec groupResourceSpec,
final int operatorManagedMemoryWeight,
final int groupManagedMemoryWeight,
final StreamConfig operatorConfig) {
- final double managedMemoryFraction;
-
if (groupResourceSpec.equals(ResourceSpec.UNKNOWN)) {
- managedMemoryFraction = groupManagedMemoryWeight > 0
- ? getFractionRoundedDown(operatorManagedMemoryWeight, groupManagedMemoryWeight)
- : 0.0;
+ operatorConfig.setManagedMemoryFraction(
+ groupManagedMemoryWeight > 0 ?
+ getFractionRoundedDown(operatorManagedMemoryWeight, groupManagedMemoryWeight) :
+ 0.0);
} else {
- final long groupManagedMemoryBytes = groupResourceSpec.getManagedMemory().getBytes();
- managedMemoryFraction = groupManagedMemoryBytes > 0
- ? getFractionRoundedDown(operatorResourceSpec.getManagedMemory().getBytes(), groupManagedMemoryBytes)
- : 0.0;
+ // Supporting for fine grained resource specs is still under developing.
+ // This branch should not be executed in production. Not throwing exception for testing purpose.
+ LOG.error("Failed setting managed memory fractions. " +
+ " Operators may not be able to use managed memory properly." +
+ " Calculating managed memory fractions with fine grained resource spec is currently not supported.");
Review comment:
I know ATM we will never go into this branch by design, but does it make sense to throw an exception here?
##########
File path: flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
##########
@@ -401,6 +403,22 @@
.withDescription("Fraction of Total Flink Memory to be used as Managed Memory, if Managed Memory size is not"
+ " explicitly specified.");
+ /**
+ * Weights of managed memory consumers.
+ */
+ // Do not advertise this option until the feature is completed.
+ @Documentation.ExcludeFromDocumentation
+ public static final ConfigOption<Map<String, String>> MANAGED_MEMORY_CONSUMER_WEIGHTS =
+ key("taskmanager.memory.managed.consumer-weights")
+ .mapType()
+ .defaultValue(new HashMap<String, String>() {{
+ put("DATAPROC", "70");
+ put("PYTHON", "30");
Review comment:
It would be good to have an `enum` which holds all supported kinds.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java
##########
@@ -49,6 +48,8 @@
@Internal
public class StreamNode implements Serializable {
+ public static final int DEFAULT_MANAGED_MEMORY_WEIGHT = 1;
Review comment:
Could be package-private.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
##########
@@ -177,11 +178,13 @@ private JobGraph createJobGraph() {
setSlotSharingAndCoLocation();
+ // For now, only consider managed memory for batch algorithms.
+ // TODO: extend managed memory fraction calculations w.r.t. various managed memory use cases.
setManagedMemoryFraction(
Collections.unmodifiableMap(jobVertices),
Collections.unmodifiableMap(vertexConfigs),
Collections.unmodifiableMap(chainedConfigs),
- id -> streamGraph.getStreamNode(id).getManagedMemoryWeight());
+ id -> streamGraph.getStreamNode(id).getManagedMemoryUseCaseWeights().getOrDefault(ManagedMemoryUseCase.BATCH_OP, 0));
Review comment:
We probably need a static variable for the default weight.
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java
##########
@@ -557,10 +557,11 @@ public void testSetManagedMemoryWeight() {
final StreamGraph streamGraph = env.getStreamGraph();
for (StreamNode streamNode : streamGraph.getStreamNodes()) {
- final int expectedWeight = streamNode.getOperatorName().contains("source")
- ? 123
- : StreamNode.DEFAULT_MANAGED_MEMORY_WEIGHT;
- assertEquals(expectedWeight, streamNode.getManagedMemoryWeight());
+ if (streamNode.getOperatorName().contains("source")) {
+ assertThat(streamNode.getManagedMemoryUseCaseWeights().get(ManagedMemoryUseCase.BATCH_OP), is(123));
+ } else {
+ assertThat(streamNode.getManagedMemoryUseCaseWeights().size(), is(0));
Review comment:
Then, the expected value should be a reference instead of a specific value.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13397:
URL: https://github.com/apache/flink/pull/13397#issuecomment-693261990
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6550",
"triggerID" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"triggerType" : "PUSH"
}, {
"hash" : "b8e95c022986da9f0b0d1b18e205f7cbc05f51ab",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6578",
"triggerID" : "b8e95c022986da9f0b0d1b18e205f7cbc05f51ab",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b8e95c022986da9f0b0d1b18e205f7cbc05f51ab Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6578)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on a change in pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
xintongsong commented on a change in pull request #13397:
URL: https://github.com/apache/flink/pull/13397#discussion_r489890377
##########
File path: flink-core/src/main/java/org/apache/flink/api/dag/Transformation.java
##########
@@ -139,12 +142,12 @@ public static int getNewNodeId() {
private ResourceSpec preferredResources = ResourceSpec.DEFAULT;
/**
- * This weight indicates how much this transformation relies on managed memory, so that
- * transformation highly relies on managed memory would be able to acquire more managed
- * memory in runtime (linear association). Note that it only works in cases of UNKNOWN
- * resources.
+ * Each entry in this map represents a use case that this transformation needs managed memory for. The key of the
+ * entry indicates the use case, while the value is the use-case-specific weight for this transformation. Managed
+ * memory reserved for an OPERATOR scope use case will be shared by all the declaring transformations within a slot
+ * according to this weight. For SLOT scope use cases, the weights are ignored.
*/
- private int managedMemoryWeight = DEFAULT_MANAGED_MEMORY_WEIGHT;
+ private final Map<ManagedMemoryUseCase, Integer> managedMemoryUseCaseWeights = new HashMap<>();
Review comment:
`ROCKSDB` and `PYTHON` are legal. The map accepts both `OPERATOR` and `SLOT` scope use cases. But only for `OPERATOR` scope use cases do the weights (values of the map) matter.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13397:
URL: https://github.com/apache/flink/pull/13397#issuecomment-693261990
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1e0209f0df58ecdc3460d426262fc89d035b83f6 UNKNOWN
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] KarmaGYZ commented on a change in pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
KarmaGYZ commented on a change in pull request #13397:
URL: https://github.com/apache/flink/pull/13397#discussion_r489294355
##########
File path: flink-core/src/main/java/org/apache/flink/api/dag/Transformation.java
##########
@@ -139,12 +142,12 @@ public static int getNewNodeId() {
private ResourceSpec preferredResources = ResourceSpec.DEFAULT;
/**
- * This weight indicates how much this transformation relies on managed memory, so that
- * transformation highly relies on managed memory would be able to acquire more managed
- * memory in runtime (linear association). Note that it only works in cases of UNKNOWN
- * resources.
+ * Each entry in this map represents a use case that this transformation needs managed memory for. The key of the
+ * entry indicates the use case, while the value is the use-case-specific weight for this transformation. Managed
+ * memory reserved for an OPERATOR scope use case will be shared by all the declaring transformations within a slot
+ * according to this weight. For SLOT scope use cases, the weights are ignored.
*/
- private int managedMemoryWeight = DEFAULT_MANAGED_MEMORY_WEIGHT;
+ private final Map<ManagedMemoryUseCase, Integer> managedMemoryUseCaseWeights = new HashMap<>();
Review comment:
Just trying to understand, is it illegal to put `ROCKSDB` or `PYTHON` in this map in your design? If so, should we check the sanity if someone tries to put `ROCKSDB` or `PYTHON` in?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on a change in pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
xintongsong commented on a change in pull request #13397:
URL: https://github.com/apache/flink/pull/13397#discussion_r489890553
##########
File path: flink-core/src/main/java/org/apache/flink/api/dag/Transformation.java
##########
@@ -139,12 +142,12 @@ public static int getNewNodeId() {
private ResourceSpec preferredResources = ResourceSpec.DEFAULT;
/**
- * This weight indicates how much this transformation relies on managed memory, so that
- * transformation highly relies on managed memory would be able to acquire more managed
- * memory in runtime (linear association). Note that it only works in cases of UNKNOWN
- * resources.
+ * Each entry in this map represents a use case that this transformation needs managed memory for. The key of the
+ * entry indicates the use case, while the value is the use-case-specific weight for this transformation. Managed
+ * memory reserved for an OPERATOR scope use case will be shared by all the declaring transformations within a slot
+ * according to this weight. For SLOT scope use cases, the weights are ignored.
*/
- private int managedMemoryWeight = DEFAULT_MANAGED_MEMORY_WEIGHT;
+ private final Map<ManagedMemoryUseCase, Integer> managedMemoryUseCaseWeights = new HashMap<>();
Review comment:
I'll try to clarify a bit in the JavaDocs.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13397:
URL: https://github.com/apache/flink/pull/13397#issuecomment-693261990
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6550",
"triggerID" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"triggerType" : "PUSH"
}, {
"hash" : "b8e95c022986da9f0b0d1b18e205f7cbc05f51ab",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "b8e95c022986da9f0b0d1b18e205f7cbc05f51ab",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1e0209f0df58ecdc3460d426262fc89d035b83f6 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6550)
* b8e95c022986da9f0b0d1b18e205f7cbc05f51ab UNKNOWN
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13397:
URL: https://github.com/apache/flink/pull/13397#issuecomment-693261990
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6550",
"triggerID" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1e0209f0df58ecdc3460d426262fc89d035b83f6 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6550)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
xintongsong commented on pull request #13397:
URL: https://github.com/apache/flink/pull/13397#issuecomment-694667340
Offline discussed with @KarmaGYZ, we decided to combine the changes of FLINK-19178 & FLINK-19179 in #13416, to improve readability and avoid leaving intermediate commits in git history.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on a change in pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
xintongsong commented on a change in pull request #13397:
URL: https://github.com/apache/flink/pull/13397#discussion_r489903233
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java
##########
@@ -49,6 +48,8 @@
@Internal
public class StreamNode implements Serializable {
+ public static final int DEFAULT_MANAGED_MEMORY_WEIGHT = 1;
Review comment:
This does not really matter since the constant is soon removed in the next commit.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong commented on pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
xintongsong commented on pull request #13397:
URL: https://github.com/apache/flink/pull/13397#issuecomment-693780307
Thanks for the review, @KarmaGYZ. Comments addressed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] xintongsong closed pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
xintongsong closed pull request #13397:
URL: https://github.com/apache/flink/pull/13397
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13397: [FLINK-19178][runtime] Extend managed memory weight/fraction interfaces for various use cases.
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13397:
URL: https://github.com/apache/flink/pull/13397#issuecomment-693261990
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6550",
"triggerID" : "1e0209f0df58ecdc3460d426262fc89d035b83f6",
"triggerType" : "PUSH"
}, {
"hash" : "b8e95c022986da9f0b0d1b18e205f7cbc05f51ab",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6578",
"triggerID" : "b8e95c022986da9f0b0d1b18e205f7cbc05f51ab",
"triggerType" : "PUSH"
}, {
"hash" : "7c2c20f97637a965be7c74243592f1157d1ac9f7",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6583",
"triggerID" : "7c2c20f97637a965be7c74243592f1157d1ac9f7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7c2c20f97637a965be7c74243592f1157d1ac9f7 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6583)
<details>
<summary>Bot commands</summary>
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
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org