You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2019/12/12 13:41:27 UTC
[flink] branch release-1.10 updated (90f46be -> d9eefeb)
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a change to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 90f46be [FLINK-15163][build] Set japicmp.referenceVersion to 1.9.1
new 952a880 [FLINK-15082] Respect new FLIP-49 taskmanager.memory.total-process.size in Mesos RM
new d9eefeb [FLINK-15082] Deprecate mesos.resourcemanager.tasks.mem in favour of taskmanager.memory.total-process.size
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../mesos_task_manager_configuration.html | 6 ----
.../MesosTaskManagerParameters.java | 31 +++++++++++++++++-
.../MesosTaskManagerParametersTest.java | 38 +++++++++++++++++++++-
3 files changed, 67 insertions(+), 8 deletions(-)
[flink] 02/02: [FLINK-15082] Deprecate
mesos.resourcemanager.tasks.mem in favour of
taskmanager.memory.total-process.size
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d9eefeb0baa4c78202c116084bc39fddc322cc52
Author: Andrey Zagrebin <az...@apache.org>
AuthorDate: Wed Dec 11 13:26:58 2019 +0100
[FLINK-15082] Deprecate mesos.resourcemanager.tasks.mem in favour of taskmanager.memory.total-process.size
This closes #10513.
---
docs/_includes/generated/mesos_task_manager_configuration.html | 6 ------
.../mesos/runtime/clusterframework/MesosTaskManagerParameters.java | 6 ++++++
2 files changed, 6 insertions(+), 6 deletions(-)
diff --git a/docs/_includes/generated/mesos_task_manager_configuration.html b/docs/_includes/generated/mesos_task_manager_configuration.html
index 6351b61..aa24d11 100644
--- a/docs/_includes/generated/mesos_task_manager_configuration.html
+++ b/docs/_includes/generated/mesos_task_manager_configuration.html
@@ -75,12 +75,6 @@
<td>Optional value to define the TaskManager’s hostname. The pattern _TASK_ is replaced by the actual id of the Mesos task. This can be used to configure the TaskManager to use Mesos DNS (e.g. _TASK_.flink-service.mesos) for name lookups.</td>
</tr>
<tr>
- <td><h5>mesos.resourcemanager.tasks.mem</h5></td>
- <td style="word-wrap: break-word;">1024</td>
- <td>Integer</td>
- <td>Memory to assign to the Mesos workers in MB.</td>
- </tr>
- <tr>
<td><h5>mesos.resourcemanager.tasks.taskmanager-cmd</h5></td>
<td style="word-wrap: break-word;">"$FLINK_HOME/bin/mesos-taskmanager.sh"</td>
<td>String</td>
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParameters.java b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParameters.java
index 6f1ac2d..bff126d 100644
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParameters.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParameters.java
@@ -57,6 +57,12 @@ public class MesosTaskManagerParameters {
public static final ConfigOption<Integer> MESOS_RM_TASKS_SLOTS =
TaskManagerOptions.NUM_TASK_SLOTS;
+ /**
+ * Total task executor container memory in megabytes to allocate.
+ *
+ * @deprecated set explicitly {@link TaskManagerOptions#TOTAL_PROCESS_MEMORY} instead
+ */
+ @Deprecated
public static final ConfigOption<Integer> MESOS_RM_TASKS_MEMORY_MB =
key("mesos.resourcemanager.tasks.mem")
.defaultValue(1024)
[flink] 01/02: [FLINK-15082] Respect new FLIP-49
taskmanager.memory.total-process.size in Mesos RM
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 952a880b70cd13fe22ca8397611168e9e55d4823
Author: Andrey Zagrebin <az...@apache.org>
AuthorDate: Tue Dec 10 09:30:58 2019 +0100
[FLINK-15082] Respect new FLIP-49 taskmanager.memory.total-process.size in Mesos RM
Usage of noew option is consistent with the legacy Mesos specific one 'mesos.resourcemanager.tasks.mem':
- If taskmanager.memory.total-process.size and mesos.resourcemanager.tasks.mem are both set and differ in their values, an exception should be thrown
- If only taskmanager.memory.total-process.size is set and mesos.resourcemanager.tasks.mem is not set, then the value configured by the former should be respected
- If only mesos.resourcemanager.tasks.mem is set and taskmanager.memory.total-process.size is not set, then the value configured by the former should be respected
---
.../MesosTaskManagerParameters.java | 25 +++++++++++++-
.../MesosTaskManagerParametersTest.java | 38 +++++++++++++++++++++-
2 files changed, 61 insertions(+), 2 deletions(-)
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParameters.java b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParameters.java
index 656be8e..6f1ac2d 100644
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParameters.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParameters.java
@@ -404,7 +404,7 @@ public class MesosTaskManagerParameters {
private static ContaineredTaskManagerParameters createContaineredTaskManagerParameters(final Configuration flinkConfig) {
double cpus = getCpuCores(flinkConfig);
- MemorySize totalProcessMemory = MemorySize.parse(flinkConfig.getInteger(MESOS_RM_TASKS_MEMORY_MB) + "m");
+ MemorySize totalProcessMemory = getTotalProcessMemory(flinkConfig);
TaskExecutorResourceSpec taskExecutorResourceSpec = TaskExecutorResourceUtils
.newResourceSpecBuilder(flinkConfig)
.withCpuCores(cpus)
@@ -422,6 +422,29 @@ public class MesosTaskManagerParameters {
return TaskExecutorResourceUtils.getCpuCoresWithFallback(configuration, fallback).getValue().doubleValue();
}
+ private static MemorySize getTotalProcessMemory(final Configuration configuration) {
+ MemorySize legacyTotalProcessMemory = MemorySize.parse(configuration.getInteger(MESOS_RM_TASKS_MEMORY_MB) + "m");
+ MemorySize unifiedTotalProcessMemory = MemorySize.parse(configuration.getString(TaskManagerOptions.TOTAL_PROCESS_MEMORY, "0"));
+
+ if (configuration.contains(MESOS_RM_TASKS_MEMORY_MB) &&
+ configuration.contains(TaskManagerOptions.TOTAL_PROCESS_MEMORY) &&
+ !legacyTotalProcessMemory.equals(unifiedTotalProcessMemory)) {
+
+ throw new IllegalConfigurationException(String.format(
+ "Inconsistent worker memory configuration: both legacy Mesos specific and the newer unified options " +
+ "are configured but they differ - %s: %d Mb (%d bytes), %s: %d Mb (%d bytes)",
+ MESOS_RM_TASKS_MEMORY_MB.key(),
+ legacyTotalProcessMemory.getMebiBytes(),
+ legacyTotalProcessMemory.getBytes(),
+ TaskManagerOptions.TOTAL_PROCESS_MEMORY.key(),
+ unifiedTotalProcessMemory.getMebiBytes(),
+ unifiedTotalProcessMemory.getBytes()));
+ }
+
+ return configuration.contains(TaskManagerOptions.TOTAL_PROCESS_MEMORY) ?
+ unifiedTotalProcessMemory : legacyTotalProcessMemory;
+ }
+
private static List<ConstraintEvaluator> parseConstraints(String mesosConstraints) {
if (mesosConstraints == null || mesosConstraints.isEmpty()) {
diff --git a/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParametersTest.java b/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParametersTest.java
index f60c658..e18d11d 100644
--- a/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParametersTest.java
+++ b/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParametersTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.mesos.runtime.clusterframework;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.IllegalConfigurationException;
+import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.util.TestLogger;
@@ -42,6 +43,9 @@ import static org.junit.Assert.assertThat;
* Tests for the {@link MesosTaskManagerParameters}.
*/
public class MesosTaskManagerParametersTest extends TestLogger {
+ private static final int TOTAL_PROCESS_MEMORY_MB = 1280;
+ private static final String TOTAL_PROCESS_MEMORY_MB_STRING = TOTAL_PROCESS_MEMORY_MB + "m";
+ private static final MemorySize TOTAL_PROCESS_MEMORY_SIZE = MemorySize.parse(TOTAL_PROCESS_MEMORY_MB_STRING);
@Test
public void testBuildVolumes() throws Exception {
@@ -233,9 +237,41 @@ public class MesosTaskManagerParametersTest extends TestLogger {
assertThat(mesosTaskManagerParameters.cpus(), is(1.5));
}
+ @Test
+ public void testUnifiedTotalProcessMemoryConfiguration() {
+ assertTotalProcessMemory(MesosTaskManagerParameters.create(getConfiguration()));
+ }
+
+ @Test
+ public void testLegacyMesosSpecificTotalProcessMemoryConfiguration() {
+ Configuration config = new Configuration();
+ config.setInteger(MESOS_RM_TASKS_MEMORY_MB, TOTAL_PROCESS_MEMORY_MB);
+ assertTotalProcessMemory(MesosTaskManagerParameters.create(config));
+ }
+
+ @Test
+ public void testUnifiedAndLegacyMesosSpecificTotalProcessMemoryConfigMatchIsOk() {
+ Configuration config = getConfiguration();
+ config.setInteger(MESOS_RM_TASKS_MEMORY_MB, TOTAL_PROCESS_MEMORY_MB);
+ assertTotalProcessMemory(MesosTaskManagerParameters.create(config));
+ }
+
+ @Test(expected = IllegalConfigurationException.class)
+ public void testUnifiedAndLegacyMesosSpecificTotalProcessMemoryConfigDifferFails() {
+ Configuration config = getConfiguration();
+ config.setInteger(MESOS_RM_TASKS_MEMORY_MB, TOTAL_PROCESS_MEMORY_MB / 2);
+ assertTotalProcessMemory(MesosTaskManagerParameters.create(config));
+ }
+
+ private void assertTotalProcessMemory(MesosTaskManagerParameters mesosTaskManagerParameters) {
+ assertThat(
+ mesosTaskManagerParameters.containeredParameters().getTaskExecutorResourceSpec().getTotalProcessMemorySize(),
+ is(TOTAL_PROCESS_MEMORY_SIZE));
+ }
+
private static Configuration getConfiguration() {
Configuration config = new Configuration();
- config.setInteger(MESOS_RM_TASKS_MEMORY_MB, 1280);
+ config.setString(TaskManagerOptions.TOTAL_PROCESS_MEMORY, TOTAL_PROCESS_MEMORY_MB_STRING);
return config;
}