[Fix-17389][Http-Task] Fix json parsing exception in request body

dolphinscheduler-task-plugin/dolphinscheduler-task-http/src/main/java/org/apache/dolphinscheduler/plugin/task/http/HttpTask.java

@@ -41,6 +41,9 @@
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+
 @Slf4j
 public class HttpTask extends AbstractTask {
 
@@ -206,7 +209,7 @@
                 .filter(httpProperty -> httpProperty.getHttpParametersType() != null)
                 .filter(httpProperty -> httpProperty.getHttpParametersType().equals(HttpParametersType.HEADERS)
                         && !httpProperty.getProp().equalsIgnoreCase(HttpConstants.CONTENT_TYPE))
                 .peek((httpProperty) -> {
                     httpProperty.setProp(ParameterUtils.convertParameterPlaceholders(httpProperty.getProp(),
                             ParameterUtils.convert(taskExecutionContext.getPrepareParamsMap())));
                     httpProperty.setValue(ParameterUtils.convertParameterPlaceholders(httpProperty.getValue(),
@@ -239,7 +242,7 @@
 
         return httpParameters.getHttpRequestParams().stream()
                 .filter(httpProperty -> httpProperty.getHttpParametersType().equals(HttpParametersType.PARAMETER))
                 .peek((httpProperty) -> {
                     httpProperty.setProp(ParameterUtils.convertParameterPlaceholders(httpProperty.getProp(),
                             ParameterUtils.convert(taskExecutionContext.getPrepareParamsMap())));
                     httpProperty.setValue(ParameterUtils.convertParameterPlaceholders(httpProperty.getValue(),
@@ -251,13 +254,18 @@
     private Map<String, Object> getRequestBody() {
         String convertedParams = ParameterUtils.convertParameterPlaceholders(httpParameters.getHttpRequestBody(),
                 ParameterUtils.convert(taskExecutionContext.getPrepareParamsMap()));
-        Map<String, String> requestBody = JSONUtils.toMap(convertedParams);
-        if (requestBody == null) {
+        JsonNode requestBodyJsonNode = JSONUtils.parseObject(convertedParams, JsonNode.class);
+        if (requestBodyJsonNode == null) {
             return null;
         }
 
-        return requestBody.entrySet().stream()
-                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+        if (!requestBodyJsonNode.isObject()) {
+            throw new IllegalArgumentException(String.format("Http request body should be a json object, but got: %s",
+                    convertedParams));
+        }
+
+        return JSONUtils.parseObject(requestBodyJsonNode.toString(), new TypeReference<Map<String, Object>>() {
+        });
     }
 
     @Override

dolphinscheduler-task-plugin/dolphinscheduler-task-http/src/test/java/org/apache/dolphinscheduler/plugin/task/http/HttpTaskTest.java

@@ -34,6 +34,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 import okhttp3.mockwebserver.Dispatcher;
 import okhttp3.mockwebserver.MockResponse;
@@ -145,6 +146,25 @@
         Assertions.assertEquals(EXIT_CODE_SUCCESS, httpTask.getExitStatusCode());
     }
 
+    @Test
+    public void testHandleWithComplexHttpBody() throws Exception {
+        String httpBody = "{\"name\":\"tom\",\"scores\":[100,98],\"metadata\":{\"grade\":\"A\"}}";
+        String httpResponse = "{\"status\": \"success\"}";
+        String url = withMockWebServer(DEFAULT_MOCK_PATH, HttpStatus.SC_OK, httpResponse);
+        HttpTask httpTask = generateHttpTask(url, HttpRequestMethod.POST, httpBody,
+                new ArrayList<>(), null, HttpCheckCondition.STATUS_CODE_DEFAULT, "");
+
+        httpTask.handle(null);
+
+        MockWebServer server = getLatestMockWebServer();
+        RecordedRequest recordedRequest = server.takeRequest(1, TimeUnit.SECONDS);
+        Assertions.assertNotNull(recordedRequest);
+        String actualRequestBody = recordedRequest.getBody().readUtf8();
+        Assertions.assertTrue(actualRequestBody.contains("\"scores\":[100,98]"));
+        Assertions.assertTrue(actualRequestBody.contains("\"metadata\":{\"grade\":\"A\"}"));
+        Assertions.assertEquals(EXIT_CODE_SUCCESS, httpTask.getExitStatusCode());
+    }
+
     @Test
     public void testHandleWithHttpParameterParams() throws Exception {
         List<HttpProperty> httpParams = new ArrayList<>();
@@ -241,11 +261,23 @@
                                       String condition, int actualResponseCode,
                                       String actualResponseBody) throws IOException {
         String url = withMockWebServer(mockPath, actualResponseCode, actualResponseBody);
+        return generateHttpTask(url, httpRequestMethod, httpBody, httpParams, prepareParamsMap,
+                httpCheckConditionType, condition);
+    }
+
+    private HttpTask generateHttpTask(String url, HttpRequestMethod httpRequestMethod, String httpBody,
+                                      List<HttpProperty> httpParams,
+                                      Map<String, String> prepareParamsMap, HttpCheckCondition httpCheckConditionType,
+                                      String condition) throws JsonProcessingException {
         String paramData =
                 generateHttpParameters(url, httpRequestMethod, httpBody, httpParams, httpCheckConditionType, condition);
         return generateHttpTaskFromParamData(paramData, prepareParamsMap);
     }
 
+    private MockWebServer getLatestMockWebServer() {
+        return mockWebServers.get(mockWebServers.size() - 1);
+    }
+
     private HttpTask generateHttpTaskFromParamData(String paramData, Map<String, String> prepareParamsMap) {
         TaskExecutionContext taskExecutionContext = Mockito.mock(TaskExecutionContext.class);
         Mockito.when(taskExecutionContext.getTaskParams()).thenReturn(paramData);