diff --git a/docs/layouts/shortcodes/generated/kubernetes_config_configuration.html b/docs/layouts/shortcodes/generated/kubernetes_config_configuration.html index a3084585bf4cba..59d290bb131313 100644 --- a/docs/layouts/shortcodes/generated/kubernetes_config_configuration.html +++ b/docs/layouts/shortcodes/generated/kubernetes_config_configuration.html @@ -176,6 +176,12 @@ Double The limit factor of memory used by job manager. The resources limit memory will be set to memory * limit-factor. + +
kubernetes.jobmanager.memory.request-factor
+ 1.0 + Double + The request factor of memory used by job manager. The resources request memory will be set to memory * request-factor. The value must be greater than 0 and less than or equal to 1. +
kubernetes.jobmanager.node-selector
(none) @@ -314,6 +320,12 @@ Double The limit factor of memory used by task manager. The resources limit memory will be set to memory * limit-factor. + +
kubernetes.taskmanager.memory.request-factor
+ 1.0 + Double + The request factor of memory used by task manager. The resources request memory will be set to memory * request-factor. The value must be greater than 0 and less than or equal to 1. +
kubernetes.taskmanager.node-selector
(none) diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java index bac5cced799016..46501139701018 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java @@ -158,6 +158,15 @@ public class KubernetesConfigOptions { "The limit factor of cpu used by job manager. " + "The resources limit cpu will be set to cpu * limit-factor."); + public static final ConfigOption JOB_MANAGER_MEMORY_REQUEST_FACTOR = + key("kubernetes.jobmanager.memory.request-factor") + .doubleType() + .defaultValue(1.0) + .withDescription( + "The request factor of memory used by job manager. " + + "The resources request memory will be set to memory * request-factor. " + + "The value must be greater than 0 and less than or equal to 1."); + public static final ConfigOption JOB_MANAGER_MEMORY_LIMIT_FACTOR = key("kubernetes.jobmanager.memory.limit-factor") .doubleType() @@ -183,6 +192,15 @@ public class KubernetesConfigOptions { "The limit factor of cpu used by task manager. " + "The resources limit cpu will be set to cpu * limit-factor."); + public static final ConfigOption TASK_MANAGER_MEMORY_REQUEST_FACTOR = + key("kubernetes.taskmanager.memory.request-factor") + .doubleType() + .defaultValue(1.0) + .withDescription( + "The request factor of memory used by task manager. " + + "The resources request memory will be set to memory * request-factor. " + + "The value must be greater than 0 and less than or equal to 1."); + public static final ConfigOption TASK_MANAGER_MEMORY_LIMIT_FACTOR = key("kubernetes.taskmanager.memory.limit-factor") .doubleType() diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecorator.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecorator.java index 00977ef89447a7..fd666de17b0535 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecorator.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecorator.java @@ -149,6 +149,7 @@ private Container decorateMainContainer(Container container) { KubernetesUtils.getResourceRequirements( requirementsInPodTemplate, kubernetesJobManagerParameters.getJobManagerMemoryMB(), + kubernetesJobManagerParameters.getJobManagerMemoryRequestFactor(), kubernetesJobManagerParameters.getJobManagerMemoryLimitFactor(), kubernetesJobManagerParameters.getJobManagerCPU(), kubernetesJobManagerParameters.getJobManagerCPULimitFactor(), diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecorator.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecorator.java index e9947f5bb4a028..5a709e5405c7b2 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecorator.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecorator.java @@ -162,6 +162,7 @@ private Container decorateMainContainer(Container container) { KubernetesUtils.getResourceRequirements( requirementsInPodTemplate, kubernetesTaskManagerParameters.getTaskManagerMemoryMB(), + kubernetesTaskManagerParameters.getTaskManagerMemoryRequestFactor(), kubernetesTaskManagerParameters.getTaskManagerMemoryLimitFactor(), kubernetesTaskManagerParameters.getTaskManagerCPU(), kubernetesTaskManagerParameters.getTaskManagerCPULimitFactor(), diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParameters.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParameters.java index 2b7b4575ebc006..95e7dc7343103f 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParameters.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParameters.java @@ -154,6 +154,16 @@ public double getJobManagerCPULimitFactor() { return limitFactor; } + public double getJobManagerMemoryRequestFactor() { + final double requestFactor = + flinkConfig.get(KubernetesConfigOptions.JOB_MANAGER_MEMORY_REQUEST_FACTOR); + checkArgument( + requestFactor > 0 && requestFactor <= 1, + "%s should be greater than 0 and less than or equal to 1.", + KubernetesConfigOptions.JOB_MANAGER_MEMORY_REQUEST_FACTOR.key()); + return requestFactor; + } + public double getJobManagerMemoryLimitFactor() { final double limitFactor = flinkConfig.get(KubernetesConfigOptions.JOB_MANAGER_MEMORY_LIMIT_FACTOR); diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesTaskManagerParameters.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesTaskManagerParameters.java index f9271d8b44c9c2..5354e5e0ebcde6 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesTaskManagerParameters.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesTaskManagerParameters.java @@ -142,6 +142,16 @@ public double getTaskManagerCPULimitFactor() { return limitFactor; } + public double getTaskManagerMemoryRequestFactor() { + final double requestFactor = + flinkConfig.get(KubernetesConfigOptions.TASK_MANAGER_MEMORY_REQUEST_FACTOR); + checkArgument( + requestFactor > 0 && requestFactor <= 1, + "%s should be greater than 0 and less than or equal to 1.", + KubernetesConfigOptions.TASK_MANAGER_MEMORY_REQUEST_FACTOR.key()); + return requestFactor; + } + public double getTaskManagerMemoryLimitFactor() { final double limitFactor = flinkConfig.get(KubernetesConfigOptions.TASK_MANAGER_MEMORY_LIMIT_FACTOR); diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java index 7786c0add56487..93893712500d94 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java @@ -409,6 +409,7 @@ public static CompletedCheckpointStore createCompletedCheckpointStore( * * @param resourceRequirements resource requirements in pod template * @param mem Memory in mb. + * @param memoryRequestFactor request factor for the memory, used to set the request resources. * @param memoryLimitFactor limit factor for the memory, used to set the limit resources. * @param cpu cpu. * @param cpuLimitFactor limit factor for the cpu, used to set the limit resources. @@ -419,6 +420,7 @@ public static CompletedCheckpointStore createCompletedCheckpointStore( public static ResourceRequirements getResourceRequirements( ResourceRequirements resourceRequirements, int mem, + double memoryRequestFactor, double memoryLimitFactor, double cpu, double cpuLimitFactor, @@ -426,13 +428,14 @@ public static ResourceRequirements getResourceRequirements( Map externalResourceConfigKeys) { final Quantity cpuQuantity = new Quantity(String.valueOf(cpu)); final Quantity cpuLimitQuantity = new Quantity(String.valueOf(cpu * cpuLimitFactor)); - final Quantity memQuantity = new Quantity(mem + Constants.RESOURCE_UNIT_MB); + final Quantity memQuantityRequest = + new Quantity(((int) (mem * memoryRequestFactor)) + Constants.RESOURCE_UNIT_MB); final Quantity memQuantityLimit = new Quantity(((int) (mem * memoryLimitFactor)) + Constants.RESOURCE_UNIT_MB); ResourceRequirementsBuilder resourceRequirementsBuilder = new ResourceRequirementsBuilder(resourceRequirements) - .addToRequests(Constants.RESOURCE_NAME_MEMORY, memQuantity) + .addToRequests(Constants.RESOURCE_NAME_MEMORY, memQuantityRequest) .addToRequests(Constants.RESOURCE_NAME_CPU, cpuQuantity) .addToLimits(Constants.RESOURCE_NAME_MEMORY, memQuantityLimit) .addToLimits(Constants.RESOURCE_NAME_CPU, cpuLimitQuantity); diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecoratorTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecoratorTest.java index 99fccde7c85043..de34c74d7d091b 100644 --- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecoratorTest.java +++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecoratorTest.java @@ -61,6 +61,7 @@ class InitJobManagerDecoratorTest extends KubernetesJobManagerTestBase { Arrays.asList( new Toleration("NoSchedule", "key1", "Equal", null, "value1"), new Toleration("NoExecute", "key2", "Exists", 6000L, null)); + private static final double JOB_MANAGER_MEMORY_REQUEST_FACTOR = 0.5; private static final String USER_DEFINED_FLINK_LOG_DIR = "/path/of/flink-log"; @@ -78,6 +79,9 @@ protected void setupFlinkConfig() { this.flinkConfig.set(KubernetesConfigOptions.JOB_MANAGER_ANNOTATIONS, ANNOTATIONS); this.flinkConfig.setString( KubernetesConfigOptions.JOB_MANAGER_TOLERATIONS.key(), TOLERATION_STRING); + this.flinkConfig.set( + KubernetesConfigOptions.JOB_MANAGER_MEMORY_REQUEST_FACTOR, + JOB_MANAGER_MEMORY_REQUEST_FACTOR); this.flinkConfig.set(KubernetesConfigOptions.FLINK_LOG_DIR, USER_DEFINED_FLINK_LOG_DIR); } @@ -117,7 +121,9 @@ void testMainContainerResourceRequirements() { final Map requests = resourceRequirements.getRequests(); assertThat(requests.get("cpu").getAmount()).isEqualTo(Double.toString(JOB_MANAGER_CPU)); assertThat(requests.get("memory").getAmount()) - .isEqualTo(String.valueOf(JOB_MANAGER_MEMORY)); + .isEqualTo( + Integer.toString( + (int) (JOB_MANAGER_MEMORY * JOB_MANAGER_MEMORY_REQUEST_FACTOR))); final Map limits = resourceRequirements.getLimits(); assertThat(limits.get("cpu").getAmount()) diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecoratorTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecoratorTest.java index 4caab5c57bf649..eb32b47d53cd46 100644 --- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecoratorTest.java +++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecoratorTest.java @@ -71,6 +71,7 @@ class InitTaskManagerDecoratorTest extends KubernetesTaskManagerTestBase { private static final String RESOURCE_NAME = "test"; private static final Long RESOURCE_AMOUNT = 2L; private static final String RESOURCE_CONFIG_KEY = "test.com/test"; + private static final double TASK_MANAGER_MEMORY_REQUEST_FACTOR = 0.5; private static final String USER_DEFINED_FLINK_LOG_DIR = "/path/of/flink-log"; @@ -88,6 +89,9 @@ protected void setupFlinkConfig() { this.flinkConfig.set(KubernetesConfigOptions.TASK_MANAGER_ANNOTATIONS, ANNOTATIONS); this.flinkConfig.setString( KubernetesConfigOptions.TASK_MANAGER_TOLERATIONS.key(), TOLERATION_STRING); + this.flinkConfig.set( + KubernetesConfigOptions.TASK_MANAGER_MEMORY_REQUEST_FACTOR, + TASK_MANAGER_MEMORY_REQUEST_FACTOR); // Set up external resource configs flinkConfig.setString(ExternalResourceOptions.EXTERNAL_RESOURCE_LIST.key(), RESOURCE_NAME); @@ -146,7 +150,9 @@ void testMainContainerResourceRequirements() { final Map requests = resourceRequirements.getRequests(); assertThat(requests.get("cpu").getAmount()).isEqualTo(Double.toString(TASK_MANAGER_CPU)); assertThat(requests.get("memory").getAmount()) - .isEqualTo(String.valueOf(TOTAL_PROCESS_MEMORY)); + .isEqualTo( + Integer.toString( + (int) (TOTAL_PROCESS_MEMORY * TASK_MANAGER_MEMORY_REQUEST_FACTOR))); final Map limits = resourceRequirements.getLimits(); assertThat(limits.get("cpu").getAmount()) diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParametersTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParametersTest.java index 24b1ce1af0a84a..8d79df9a601d76 100644 --- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParametersTest.java +++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParametersTest.java @@ -47,6 +47,7 @@ class KubernetesJobManagerParametersTest extends KubernetesTestBase { private static final double JOB_MANAGER_CPU = 2.0; private static final double JOB_MANAGER_CPU_LIMIT_FACTOR = 2.5; + private static final double JOB_MANAGER_MEMORY_REQUEST_FACTOR = 0.5; private static final double JOB_MANAGER_MEMORY_LIMIT_FACTOR = 2.0; private final ClusterSpecification clusterSpecification = @@ -168,6 +169,39 @@ void testGetJobManagerCPULimitFactor() { .isEqualTo(JOB_MANAGER_CPU_LIMIT_FACTOR, within(0.00001)); } + @Test + void testGetDefaultJobManagerMemoryRequestFactor() { + assertThat(kubernetesJobManagerParameters.getJobManagerMemoryRequestFactor()) + .isEqualTo(1.0, within(0.00001)); + } + + @Test + void testGetJobManagerMemoryRequestFactor() { + flinkConfig.set( + KubernetesConfigOptions.JOB_MANAGER_MEMORY_REQUEST_FACTOR, + JOB_MANAGER_MEMORY_REQUEST_FACTOR); + assertThat(kubernetesJobManagerParameters.getJobManagerMemoryRequestFactor()) + .isEqualTo(JOB_MANAGER_MEMORY_REQUEST_FACTOR, within(0.00001)); + } + + @Test + void testGetJobManagerMemoryRequestFactorShouldFailIfGreaterThanOne() { + flinkConfig.set(KubernetesConfigOptions.JOB_MANAGER_MEMORY_REQUEST_FACTOR, 1.1); + + assertThatThrownBy(kubernetesJobManagerParameters::getJobManagerMemoryRequestFactor) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("greater than 0 and less than or equal to 1"); + } + + @Test + void testGetJobManagerMemoryRequestFactorShouldFailIfLessThanOrEqualToZero() { + flinkConfig.set(KubernetesConfigOptions.JOB_MANAGER_MEMORY_REQUEST_FACTOR, 0.0); + + assertThatThrownBy(kubernetesJobManagerParameters::getJobManagerMemoryRequestFactor) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("greater than 0 and less than or equal to 1"); + } + @Test void testGetJobManagerMemoryLimitFactor() { flinkConfig.set( diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesTaskManagerParametersTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesTaskManagerParametersTest.java index 41d8e5d7dd71b0..8075a32794a1fe 100644 --- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesTaskManagerParametersTest.java +++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesTaskManagerParametersTest.java @@ -35,6 +35,7 @@ import java.util.Map; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.assertj.core.api.Assertions.within; /** General tests for the {@link KubernetesTaskManagerParameters}. */ @@ -43,6 +44,7 @@ class KubernetesTaskManagerParametersTest extends KubernetesTestBase { private static final int TASK_MANAGER_MEMORY = 1024; private static final double TASK_MANAGER_CPU = 1.2; private static final double TASK_MANAGER_CPU_LIMIT_FACTOR = 2.0; + private static final double TASK_MANAGER_MEMORY_REQUEST_FACTOR = 0.5; private static final double TASK_MANAGER_MEMORY_LIMIT_FACTOR = 2.0; private static final int RPC_PORT = 13001; @@ -150,6 +152,39 @@ void testGetTaskManagerCPULimitFactor() { .isEqualTo(TASK_MANAGER_CPU_LIMIT_FACTOR, within(0.00001)); } + @Test + void testGetDefaultTaskManagerMemoryRequestFactor() { + assertThat(kubernetesTaskManagerParameters.getTaskManagerMemoryRequestFactor()) + .isEqualTo(1.0, within(0.00001)); + } + + @Test + void testGetTaskManagerMemoryRequestFactor() { + flinkConfig.set( + KubernetesConfigOptions.TASK_MANAGER_MEMORY_REQUEST_FACTOR, + TASK_MANAGER_MEMORY_REQUEST_FACTOR); + assertThat(kubernetesTaskManagerParameters.getTaskManagerMemoryRequestFactor()) + .isEqualTo(TASK_MANAGER_MEMORY_REQUEST_FACTOR, within(0.00001)); + } + + @Test + void testGetTaskManagerMemoryRequestFactorShouldFailIfGreaterThanOne() { + flinkConfig.set(KubernetesConfigOptions.TASK_MANAGER_MEMORY_REQUEST_FACTOR, 1.1); + + assertThatThrownBy(kubernetesTaskManagerParameters::getTaskManagerMemoryRequestFactor) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("greater than 0 and less than or equal to 1"); + } + + @Test + void testGetTaskManagerMemoryRequestFactorShouldFailIfLessThanOrEqualToZero() { + flinkConfig.set(KubernetesConfigOptions.TASK_MANAGER_MEMORY_REQUEST_FACTOR, 0.0); + + assertThatThrownBy(kubernetesTaskManagerParameters::getTaskManagerMemoryRequestFactor) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("greater than 0 and less than or equal to 1"); + } + @Test void testGetTaskManagerMemoryLimitFactor() { assertThat(kubernetesTaskManagerParameters.getTaskManagerMemoryLimitFactor())