[GitHub] flink pull request #2669: [FLINK-4871] [mini cluster] Add memory calculation...

[tillrohrmann]

Github user tillrohrmann closed the pull request at:

https://github.com/apache/flink/pull/2669


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2669: [FLINK-4871] [mini cluster] Add memory calculation...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2669#discussion_r84509867
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterConfiguration.java
 ---
@@ -96,11 +101,20 @@ public void setCommonRpcBindAddress(String 
bindAddress) {
this.commonBindAddress = bindAddress;
}
 
+   public void setManagedMemoryPerTaskManager(long 
managedMemoryPerTaskManager) {
+   checkArgument(managedMemoryPerTaskManager > 0, "must have more 
than 0 MB of memory for the TaskManager.");
+   this.managedMemoryPerTaskManager = managedMemoryPerTaskManager;
+   }
+
// 

//  getters
// 

 
public Configuration getConfiguration() {
+   // update the memory in case that we've changed the number of 
components (TM, RM, JM)
+   long memory = calculateManagedMemoryPerTaskManager();
--- End diff --

After this method has been called, you can't change the memory 
configuration anymore because the config value will prevent new calculation in 
`calculateManagedMemoryPerTaskManager`. Is that desired?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2669: [FLINK-4871] [mini cluster] Add memory calculation...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2669#discussion_r84509754
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterConfiguration.java
 ---
@@ -96,11 +101,20 @@ public void setCommonRpcBindAddress(String 
bindAddress) {
this.commonBindAddress = bindAddress;
}
 
+   public void setManagedMemoryPerTaskManager(long 
managedMemoryPerTaskManager) {
+   checkArgument(managedMemoryPerTaskManager > 0, "must have more 
than 0 MB of memory for the TaskManager.");
+   this.managedMemoryPerTaskManager = managedMemoryPerTaskManager;
+   }
+
// 

//  getters
// 

 
public Configuration getConfiguration() {
+   // update the memory in case that we've changed the number of 
components (TM, RM, JM)
+   long memory = calculateManagedMemoryPerTaskManager();
--- End diff --

Getters should usually not perform any calculation. How about changing the 
method name to `updateConfiguration()`? 


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2669: [FLINK-4871] [mini cluster] Add memory calculation...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2669#discussion_r84508885
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterConfiguration.java
 ---
@@ -162,4 +180,63 @@ public String toString() {
", config=" + config +
'}';
}
+
+   /**
+* Calculate the managed memory per task manager. The memory is 
calculated in the following
+* order:
+*
+* 1. Return {@link #managedMemoryPerTaskManager} if set
+* 2. Return 
config.getInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY) if set
+* 3. Distribute the available free memory equally among all components 
(JMs, RMs and TMs) and
+* calculate the managed memory from the share of memory for a single 
task manager.
+*
+* @return
+*/
+   private long calculateManagedMemoryPerTaskManager() {
--- End diff --

`getOrCalculateManagedMemoryPerTaskManager`? 😃


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2669: [FLINK-4871] [mini cluster] Add memory calculation...

[tillrohrmann]

GitHub user tillrohrmann opened a pull request:

https://github.com/apache/flink/pull/2669

[FLINK-4871] [mini cluster] Add memory calculation for TaskManagers to 
MiniCluster

This PR is based on #2651, #2655 and #2657.

If the managed memory size for the task manager has not been set in the 
Configuration, then
it is automatically calculated by dividing the available memory by the 
number of distributed
components. Additionally this PR allows to provide a MetricRegistry to the 
TaskManagerRunner.
That way it is possible to use the MiniCluster's MetricRegistry.

Add memory calculation for task managers.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/tillrohrmann/flink miniClusterUpdate

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/flink/pull/2669.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #2669


commit 6487af737b57ca16190c0f4a6b63d4afd2af2b06
Author: Till Rohrmann 
Date:   2016-10-17T14:22:16Z

[FLINK-4847] Let RpcEndpoint.start/shutDown throw exceptions

Allowing the RpcEndpoint.start/shutDown to throw exceptions will help to 
let rpc endpoints
to quickly fail without having to use a callback like the FatalErrorHandler.

commit cf7661aafb71649360ad6159f27a82433bfd8f75
Author: Till Rohrmann 
Date:   2016-10-17T14:03:02Z

[FLINK-4851] [rm] Introduce FatalErrorHandler and MetricRegistry to RM

This PR introduces a FatalErrorHandler and the MetricRegistry to the RM. 
The FatalErrorHandler is used to handle fatal errors. Additionally, the PR adds 
the MetricRegistry to the RM which can be used
to register metrics.

Apart from these changes the PR restructures the code of the RM a little 
bit and fixes some
blocking operations.

The PR also moves the TestingFatalErrorHandler into the util package of 
flink-runtime test. That
it is usable across multiple tests.

Introduce ResourceManagerRunner to handle errors in the ResourceManager

commit 61d328adb637c44889aa724c270c39657d1289c2
Author: Till Rohrmann 
Date:   2016-10-18T16:03:00Z

[FLINK-4853] [rm] Clean up job manager registration at the resource manager

Introduce the JobLeaderIdService which automatically retrieves the current 
job leader id.
This job leader id is used to validate job manager registartion attempts. 
Additionally, it
is used to disconnect old job leaders from the resource manager.

Add comments

commit 58e8d6c06b55456793d2dffbee309e491d21d309
Author: Till Rohrmann 
Date:   2016-10-20T09:07:08Z

[FLINK-4871] [mini cluster] Add memory calculation for TaskManagers to 
MiniCluster

If the managed memory size for the task manager has not been set in the 
Configuration, then
it is automatically calculated by dividing the available memory by the 
number of distributed
components. Additionally this PR allows to provide a MetricRegistry to the 
TaskManagerRunner.
That way it is possible to use the MiniCluster's MetricRegistry.

Add memory calculation for task managers




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---