apache · ChenlingXie · Jul 9, 2026
diff --git a/docs/layouts/shortcodes/generated/kubernetes_config_configuration.html b/docs/layouts/shortcodes/generated/kubernetes_config_configuration.html
@@ -176,6 +176,12 @@
             <td>Double</td>
             <td>The limit factor of memory used by job manager. The resources limit memory will be set to memory * limit-factor.</td>
         </tr>
+        <tr>
+            <td><h5>kubernetes.jobmanager.memory.request-factor</h5></td>
+            <td style="word-wrap: break-word;">1.0</td>
+            <td>Double</td>
+            <td>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.</td>
+        </tr>
         <tr>
             <td><h5>kubernetes.jobmanager.node-selector</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
@@ -314,6 +320,12 @@
             <td>Double</td>
             <td>The limit factor of memory used by task manager. The resources limit memory will be set to memory * limit-factor.</td>
         </tr>
+        <tr>
+            <td><h5>kubernetes.taskmanager.memory.request-factor</h5></td>
+            <td style="word-wrap: break-word;">1.0</td>
+            <td>Double</td>
+            <td>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.</td>
+        </tr>
         <tr>
             <td><h5>kubernetes.taskmanager.node-selector</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

diff --git a/...etes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java b/...etes/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<Double> 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<Double> 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<Double> 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<Double> TASK_MANAGER_MEMORY_LIMIT_FACTOR =
             key("kubernetes.taskmanager.memory.limit-factor")
                     .doubleType()

diff --git a/.../main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecorator.java b/.../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/...main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecorator.java b/...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/...ava/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParameters.java b/...ava/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/...va/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesTaskManagerParameters.java b/...va/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
@@ -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,20 +420,22 @@ public static CompletedCheckpointStore createCompletedCheckpointStore(
     public static ResourceRequirements getResourceRequirements(
             ResourceRequirements resourceRequirements,
             int mem,
+            double memoryRequestFactor,
             double memoryLimitFactor,
             double cpu,
             double cpuLimitFactor,
             Map<String, ExternalResource> externalResources,
             Map<String, String> 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/...t/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecoratorTest.java b/...t/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<String, Quantity> 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<String, Quantity> limits = resourceRequirements.getLimits();
         assertThat(limits.get("cpu").getAmount())

diff --git a/.../java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecoratorTest.java b/.../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<String, Quantity> 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<String, Quantity> limits = resourceRequirements.getLimits();
         assertThat(limits.get("cpu").getAmount())

diff --git a/...org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParametersTest.java b/...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/...rg/apache/flink/kubernetes/kubeclient/parameters/KubernetesTaskManagerParametersTest.java b/...rg/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())