[Feature][RestAPI] Support submit job with seatunnel style hocon format config

seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/ClusterSeaTunnelContainer.java

@@ -215,6 +215,30 @@ public void testStartWithSavePointWithoutJobId() {
                         });
     }
 
+    @Test
+    public void testStartWithSavePointWithoutJobIdAndHoconStyle() {
+        Arrays.asList(server, secondServer)
+                .forEach(
+                        container -> {
+                            Tuple3<Integer, String, Long> task = tasks.get(0);
+                            Response response =
+                                    submitJobWithHoconStyle(
+                                            "BATCH",
+                                            container,
+                                            task._1(),
+                                            task._2(),
+                                            true,
+                                            jobName,
+                                            paramJobName);
+                            response.then()
+                                    .statusCode(400)
+                                    .body(
+                                            "message",
+                                            equalTo(
+                                                    "Please provide jobId when start with save point."));
+                        });
+    }
+
     @Test
     public void testStartWithSavePointWithoutJobIdV2() {
         Arrays.asList(server, secondServer)
@@ -239,6 +263,30 @@ public void testStartWithSavePointWithoutJobIdV2() {
                         });
     }
 
+    @Test
+    public void testStartWithSavePointWithoutJobIdV2AndHoconStyle() {
+        Arrays.asList(server, secondServer)
+                .forEach(
+                        container -> {
+                            Tuple3<Integer, String, Long> task = tasks.get(1);
+                            Response response =
+                                    submitJobWithHoconStyle(
+                                            "BATCH",
+                                            container,
+                                            task._1(),
+                                            task._2(),
+                                            true,
+                                            jobName,
+                                            paramJobName);
+                            response.then()
+                                    .statusCode(400)
+                                    .body(
+                                            "message",
+                                            equalTo(
+                                                    "Please provide jobId when start with save point."));
+                        });
+    }
+
     @Test
     public void testStopJob() {
         AtomicInteger i = new AtomicInteger();
@@ -798,6 +846,70 @@ private void submitJobs(
         return JsonUtil.toJson(jobList);
     }
 
+    private Response submitJobWithHoconStyle(
+            String jobMode,
+            GenericContainer<?> container,
+            int port,
+            String contextPath,
+            boolean isStartWithSavePoint,
+            String jobName,
+            String paramJobName) {
+        String requestBody =
+                String.format(
+                        "env {\n"
+                                + "  job.name = \"%s\"\n"
+                                + "  job.mode = \"%s\"\n"
+                                + "}\n\n"
+                                + "source {\n"
+                                + "  FakeSource {\n"
+                                + "    result_table_name = \"fake\"\n"
+                                + "    schema = {\n"
+                                + "      fields {\n"
+                                + "        name = \"string\"\n"
+                                + "        age = \"int\"\n"
+                                + "        card = \"int\"\n"
+                                + "      }\n"
+                                + "    }\n"
+                                + "  }\n"
+                                + "}\n\n"
+                                + "transform {\n"
+                                + "}\n\n"
+                                + "sink {\n"
+                                + "  Console {\n"
+                                + "    source_table_name = \"fake\"\n"
+                                + "  }\n"
+                                + "}\n",
+                        jobName, jobMode);
+        String parameters = null;
+        if (paramJobName != null) {
+            parameters = "jobName=" + paramJobName;
+        }
+        if (isStartWithSavePoint) {
+            parameters = parameters + "&isStartWithSavePoint=true";
+        }
+        parameters = parameters + "&configStyle=hocon";
+        Response response =
+                given().body(requestBody)
+                        .header("Content-Type", "text/plain; charset=utf-8")
+                        .post(
+                                parameters == null
+                                        ? http
+                                                + container.getHost()
+                                                + colon
+                                                + port
+                                                + contextPath
+                                                + RestConstant.SUBMIT_JOB_URL
+                                        : http
+                                                + container.getHost()
+                                                + colon
+                                                + port
+                                                + contextPath
+                                                + RestConstant.SUBMIT_JOB_URL
+                                                + "?"
+                                                + parameters);
+        return response;
+    }
+
     private Response submitJob(
             String jobMode,
             GenericContainer<?> container,

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestConstant.java

@@ -27,6 +27,8 @@ public class RestConstant {
 
     public static final String IS_STOP_WITH_SAVE_POINT = "isStopWithSavePoint";
 
+    public static final String ENV_CONFIG_STYLE = "configStyle";
+
     public static final String JOB_STATUS = "jobStatus";
 
     public static final String CREATE_TIME = "createTime";
@@ -45,6 +47,8 @@ public class RestConstant {
 
     public static final String METRICS = "metrics";
 
+    public static final String HOCON = "hocon";
+
     public static final String TABLE_SOURCE_RECEIVED_COUNT = "TableSourceReceivedCount";
     public static final String TABLE_SINK_WRITE_COUNT = "TableSinkWriteCount";
     public static final String TABLE_SOURCE_RECEIVED_QPS = "TableSourceReceivedQPS";

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/JobInfoService.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.engine.server.rest.service;
 
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory;
 
 import org.apache.seatunnel.api.common.metrics.JobMetrics;
 import org.apache.seatunnel.common.utils.JsonUtils;
@@ -37,11 +38,15 @@
 import com.hazelcast.spi.impl.NodeEngineImpl;
 import scala.Tuple2;
 
+import java.nio.charset.StandardCharsets;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.seatunnel.engine.server.rest.RestConstant.ENV_CONFIG_STYLE;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.HOCON;
+
 public class JobInfoService extends BaseService {
 
     public JobInfoService(NodeEngineImpl nodeEngine) {
@@ -157,9 +162,13 @@ public JsonObject submitJob(Map<String, String> requestParams, byte[] requestBod
                 && requestParams.get(RestConstant.JOB_ID) == null) {
             throw new IllegalArgumentException("Please provide jobId when start with save point.");
         }
-
-        Config config = RestUtil.buildConfig(requestHandle(requestBody), false);
-
+        Config config;
+        if (HOCON.equalsIgnoreCase(requestParams.get(ENV_CONFIG_STYLE))) {
+            String requestBodyStr = new String(requestBody, StandardCharsets.UTF_8);
+            config = ConfigFactory.parseString(requestBodyStr);
+        } else {
+            config = RestUtil.buildConfig(requestHandle(requestBody), false);
+        }
         SeaTunnelServer seaTunnelServer = getSeaTunnelServer(false);
         return submitJobInternal(config, requestParams, seaTunnelServer, nodeEngine.getNode());
     }

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/servlet/BaseServlet.java

@@ -116,6 +116,20 @@ protected byte[] requestBody(HttpServletRequest req) throws IOException {
         return requestBody.getBytes(StandardCharsets.UTF_8);
     }
 
+    public byte[] requestHoconBody(HttpServletRequest req) throws IOException {
+        StringBuilder stringBuilder = new StringBuilder();
+        String line;
+
+        try (BufferedReader reader = req.getReader()) {
+            while ((line = reader.readLine()) != null) {
+                stringBuilder.append(line).append("\n");
+            }
+        }
+
+        String requestBody = stringBuilder.toString();
+        return requestBody.getBytes(StandardCharsets.UTF_8);
+    }
+
     protected Map<String, String> getParameterMap(HttpServletRequest req) {
         Map<String, String> reqParameterMap = new HashMap<>();
 

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/servlet/SubmitJobServlet.java

@@ -27,6 +27,9 @@
 import java.io.IOException;
 import java.util.Map;
 
+import static org.apache.seatunnel.engine.server.rest.RestConstant.ENV_CONFIG_STYLE;
+import static org.apache.seatunnel.engine.server.rest.RestConstant.HOCON;
+
 public class SubmitJobServlet extends BaseServlet {
     private final JobInfoService jobInfoService;
 
@@ -39,7 +42,10 @@ public SubmitJobServlet(NodeEngineImpl nodeEngine) {
     public void doPost(HttpServletRequest req, HttpServletResponse resp) throws IOException {
 
         Map<String, String> requestParams = getParameterMap(req);
-
-        writeJson(resp, jobInfoService.submitJob(requestParams, requestBody(req)));
+        if (HOCON.equalsIgnoreCase(requestParams.get(ENV_CONFIG_STYLE))) {
+            writeJson(resp, jobInfoService.submitJob(requestParams, requestHoconBody(req)));
+        } else {
+            writeJson(resp, jobInfoService.submitJob(requestParams, requestBody(req)));
+        }
     }
 }