[Fix] Fix memory sizes config not taking effect in Yarn mode

streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/java/org/apache/streampark/flink/client/trait/FlinkClientTrait.java

@@ -329,6 +329,7 @@ private Configuration prepareConfig(SubmitRequest submitRequest) throws Exceptio
         applyCheckpointDefaults(submitRequest, flinkConfig);
         applySavepointConfig(submitRequest, flinkConfig);
         applyEnvProperties(submitRequest, flinkConfig);
+        applyAppProperties(submitRequest, flinkConfig);
         return flinkConfig;
     }
 
@@ -455,6 +456,21 @@ private void applyEnvProperties(SubmitRequest submitRequest, Configuration flink
         }
     }
 
+    private void applyAppProperties(SubmitRequest submitRequest, Configuration flinkConfig) {
+        Map<String, String> appProperties = submitRequest.appProperties();
+        if (MapUtils.isEmpty(appProperties)) {
+            return;
+        }
+        for (Map.Entry<String, String> entry : appProperties.entrySet()) {
+            logInfo(
+                "appProperties: "
+                    + entry.getKey()
+                    + " : "
+                    + entry.getValue());
+            flinkConfig.setString(entry.getKey(), entry.getValue());
+        }
+    }
+
     public abstract void setConfig(SubmitRequest submitRequest, Configuration flinkConf);
 
     public SavepointResponse triggerSavepoint(TriggerSavepointRequest savepointRequest) throws FlinkException {

streampark-flink/streampark-flink-client/streampark-flink-client-core/src/test/java/org/apache/streampark/flink/client/test/SubmitRequestTest.java

@@ -23,6 +23,7 @@
 import org.apache.streampark.common.enums.ApplicationType;
 import org.apache.streampark.common.enums.FlinkDeployMode;
 import org.apache.streampark.common.enums.FlinkJobType;
+import org.apache.streampark.common.util.DeflaterUtils;
 import org.apache.streampark.flink.client.bean.SubmitApplicationSpec;
 import org.apache.streampark.flink.client.bean.SubmitRequest;
 
@@ -134,6 +135,83 @@ void allowNonRestoredStateShouldDefaultToFalse() {
         assertThat(request.allowNonRestoredState()).isFalse();
     }
 
+    @Test
+    void appPropertiesWithMemoryConfigFromYamlConf() {
+        String yamlContent =
+            "flink.property.jobmanager.memory.process.size: 2048m\n"
+                + "flink.property.taskmanager.memory.process.size: 4096m\n";
+        String appConf = "yaml://" + DeflaterUtils.zipString(yamlContent);
+        SubmitApplicationSpec application =
+            SubmitApplicationSpec.builder()
+                .jobType(FlinkJobType.FLINK_JAR)
+                .appConf(appConf)
+                .build();
+        SubmitRequest request =
+            new SubmitRequest(
+                FLINK_VERSION,
+                FlinkDeployMode.YARN_APPLICATION,
+                Collections.emptyMap(),
+                application,
+                null,
+                null,
+                null);
+
+        assertThat(request.appProperties())
+            .containsEntry("jobmanager.memory.process.size", "2048m")
+            .containsEntry("taskmanager.memory.process.size", "4096m");
+    }
+
+    @Test
+    void appPropertiesWithMemoryConfigFromJsonConf() {
+        String jmKey = ConfigKeys.KEY_FLINK_PROPERTY_PREFIX() + "jobmanager.memory.process.size";
+        String tmKey = ConfigKeys.KEY_FLINK_PROPERTY_PREFIX() + "taskmanager.memory.process.size";
+        String appConf = "json://{\"" + jmKey + "\":\"2g\",\"" + tmKey + "\":\"4g\"}";
+        SubmitApplicationSpec application =
+            SubmitApplicationSpec.builder()
+                .jobType(FlinkJobType.FLINK_JAR)
+                .appConf(appConf)
+                .build();
+        SubmitRequest request =
+            new SubmitRequest(
+                FLINK_VERSION,
+                FlinkDeployMode.YARN_APPLICATION,
+                Collections.emptyMap(),
+                application,
+                null,
+                null,
+                null);
+
+        assertThat(request.appProperties())
+            .containsEntry("jobmanager.memory.process.size", "2g")
+            .containsEntry("taskmanager.memory.process.size", "4g");
+    }
+
+    @Test
+    void appPropertiesWithMemoryConfigFromPropertiesConf() {
+        String propertiesContent =
+            "flink.property.jobmanager.memory.process.size=1024m\n"
+                + "flink.property.taskmanager.memory.process.size=2048m\n";
+        String appConf = "prop://" + DeflaterUtils.zipString(propertiesContent);
+        SubmitApplicationSpec application =
+            SubmitApplicationSpec.builder()
+                .jobType(FlinkJobType.FLINK_JAR)
+                .appConf(appConf)
+                .build();
+        SubmitRequest request =
+            new SubmitRequest(
+                FLINK_VERSION,
+                FlinkDeployMode.YARN_APPLICATION,
+                Collections.emptyMap(),
+                application,
+                null,
+                null,
+                null);
+
+        assertThat(request.appProperties())
+            .containsEntry("jobmanager.memory.process.size", "1024m")
+            .containsEntry("taskmanager.memory.process.size", "2048m");
+    }
+
     private static SubmitRequest createRequest(FlinkJobType jobType, String appConf) {
         SubmitApplicationSpec application =
             SubmitApplicationSpec.builder()