You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by az...@apache.org on 2020/04/20 09:41:33 UTC
[flink] 09/17: [FLINK-16745][clients] Parse JobManagerProcessSpec
from Configuration into ClusterSpecification
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 23f62bd0f203a160beb60b625c97db6ee49f7660
Author: Andrey Zagrebin <az...@apache.org>
AuthorDate: Wed Apr 8 17:05:35 2020 +0300
[FLINK-16745][clients] Parse JobManagerProcessSpec from Configuration into ClusterSpecification
---
.../AbstractContainerizedClusterClientFactory.java | 10 +++++++---
.../flink/configuration/ConfigurationUtils.java | 19 -------------------
.../runtime/jobmanager/JobManagerProcessUtils.java | 8 +++++++-
3 files changed, 14 insertions(+), 23 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/AbstractContainerizedClusterClientFactory.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/AbstractContainerizedClusterClientFactory.java
index 46af058..f78cae7 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/deployment/AbstractContainerizedClusterClientFactory.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/AbstractContainerizedClusterClientFactory.java
@@ -20,9 +20,10 @@ package org.apache.flink.client.deployment;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.ConfigurationUtils;
+import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.clusterframework.TaskExecutorProcessUtils;
+import org.apache.flink.runtime.jobmanager.JobManagerProcessUtils;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -36,8 +37,11 @@ public abstract class AbstractContainerizedClusterClientFactory<ClusterID> imple
public ClusterSpecification getClusterSpecification(Configuration configuration) {
checkNotNull(configuration);
- final int jobManagerMemoryMB = ConfigurationUtils
- .getJobManagerHeapMemory(configuration)
+ final int jobManagerMemoryMB = JobManagerProcessUtils.processSpecFromConfig(
+ JobManagerProcessUtils.getConfigurationWithLegacyHeapSizeMappedToNewConfigOption(
+ configuration,
+ JobManagerOptions.TOTAL_PROCESS_MEMORY))
+ .getTotalProcessMemorySize()
.getMebiBytes();
final int taskManagerMemoryMB = TaskExecutorProcessUtils
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigurationUtils.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigurationUtils.java
index 0408a37..4fbc512 100755
--- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigurationUtils.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigurationUtils.java
@@ -42,25 +42,6 @@ public class ConfigurationUtils {
private static final String[] EMPTY = new String[0];
/**
- * Get job manager's heap memory. This method will check the new key
- * {@link JobManagerOptions#JOB_MANAGER_HEAP_MEMORY} and
- * the old key {@link JobManagerOptions#JOB_MANAGER_HEAP_MEMORY_MB} for backwards compatibility.
- *
- * @param configuration the configuration object
- * @return the memory size of job manager's heap memory.
- */
- public static MemorySize getJobManagerHeapMemory(Configuration configuration) {
- if (configuration.containsKey(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY.key())) {
- return configuration.get(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY);
- } else if (configuration.containsKey(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY_MB.key())) {
- return MemorySize.ofMebiBytes(configuration.getInteger(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY_MB));
- } else {
- //use default value
- return JobManagerOptions.JOB_MANAGER_HEAP_MEMORY.defaultValue();
- }
- }
-
- /**
* @return extracted {@link MetricOptions#SYSTEM_RESOURCE_METRICS_PROBING_INTERVAL} or {@code Optional.empty()} if
* {@link MetricOptions#SYSTEM_RESOURCE_METRICS} are disabled.
*/
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/JobManagerProcessUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/JobManagerProcessUtils.java
index 62f61bf..82c2490 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/JobManagerProcessUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/JobManagerProcessUtils.java
@@ -24,8 +24,8 @@ import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.runtime.util.config.memory.CommonProcessMemorySpec;
import org.apache.flink.runtime.util.config.memory.JvmMetaspaceAndOverheadOptions;
-import org.apache.flink.runtime.util.config.memory.MemoryBackwardsCompatibilityUtils;
import org.apache.flink.runtime.util.config.memory.LegacyMemoryOptions;
+import org.apache.flink.runtime.util.config.memory.MemoryBackwardsCompatibilityUtils;
import org.apache.flink.runtime.util.config.memory.ProcessMemoryOptions;
import org.apache.flink.runtime.util.config.memory.ProcessMemoryUtils;
import org.apache.flink.runtime.util.config.memory.jobmanager.JobManagerFlinkMemory;
@@ -78,4 +78,10 @@ public class JobManagerProcessUtils {
ConfigOption<MemorySize> configOption) {
return LEGACY_MEMORY_UTILS.getConfWithLegacyHeapSizeMappedToNewConfigOption(configuration, configOption);
}
+
+ public static JobManagerProcessSpec createDefaultJobManagerProcessSpec(int totalProcessMemoryMb) {
+ Configuration configuration = new Configuration();
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(totalProcessMemoryMb));
+ return processSpecFromConfig(configuration);
+ }
}