[Improve][SeaTunnel-Web] Unified use of JSONUtils (#214)

apache · Sep 14, 2024 · 6cd2d7c · 6cd2d7c
1 parent 1df50cf
commit 6cd2d7c
Show file tree

Hide file tree

Showing 27 changed files with 143 additions and 604 deletions.
diff --git a/...c/main/java/org/apache/seatunnel/datasource/plugin/elasticsearch/client/EsRestClient.java b/...c/main/java/org/apache/seatunnel/datasource/plugin/elasticsearch/client/EsRestClient.java
@@ -18,7 +18,6 @@
 package org.apache.seatunnel.datasource.plugin.elasticsearch.client;
 
 import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode;
-import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode;
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 
@@ -62,18 +61,16 @@ public class EsRestClient implements AutoCloseable {
 
     private final RestClient restClient;
 
-    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
-
     private EsRestClient(RestClient restClient) {
         this.restClient = restClient;
     }
 
     public static EsRestClient createInstance(Config pluginConfig) {
         try {
             List<String> hosts =
-                    OBJECT_MAPPER.readValue(
+                    JsonUtils.toList(
                             pluginConfig.getString(ElasticSearchOptionRule.HOSTS.key()),
-                            List.class);
+                            String.class);
             Optional<String> username = Optional.empty();
             Optional<String> password = Optional.empty();
             if (pluginConfig.hasPath(ElasticSearchOptionRule.USERNAME.key())) {
@@ -209,7 +206,7 @@ private static RestClientBuilder getRestClientBuilder(
                                             keystorePassword,
                                             truststorePath,
                                             truststorePassword);
-                            sslContext.ifPresent(e -> httpClientBuilder.setSSLContext(e));
+                            sslContext.ifPresent(httpClientBuilder::setSSLContext);
                         } else {
                             SSLContext sslContext =
                                     SSLContexts.custom()
@@ -233,15 +230,13 @@ public ElasticsearchClusterInfo getClusterInfo() {
         try {
             Response response = restClient.performRequest(request);
             String result = EntityUtils.toString(response.getEntity());
-            ObjectMapper objectMapper = new ObjectMapper();
-            JsonNode jsonNode = objectMapper.readTree(result);
-            JsonNode versionNode = jsonNode.get("version");
+            String version = JsonUtils.findValue(JsonUtils.stringToJsonNode(result), "version");
+            String number = JsonUtils.findValue(JsonUtils.stringToJsonNode(version), "number");
+            String distribution =
+                    JsonUtils.findValue(JsonUtils.stringToJsonNode(version), "distribution");
             return ElasticsearchClusterInfo.builder()
-                    .clusterVersion(versionNode.get("number").asText())
-                    .distribution(
-                            Optional.ofNullable(versionNode.get("distribution"))
-                                    .map(JsonNode::asText)
-                                    .orElse(null))
+                    .clusterVersion(number)
+                    .distribution(Optional.ofNullable(distribution).orElse(null))
                     .build();
         } catch (IOException e) {
             throw new ResponseException("fail to get elasticsearch version.", e);

diff --git a/...ain/java/org/apache/seatunnel/datasource/plugin/starrocks/StarRocksDataSourceChannel.java b/...ain/java/org/apache/seatunnel/datasource/plugin/starrocks/StarRocksDataSourceChannel.java
@@ -17,10 +17,9 @@
 
 package org.apache.seatunnel.datasource.plugin.starrocks;
 
-import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.ObjectMapper;
-
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.common.utils.JsonUtils;
 import org.apache.seatunnel.datasource.plugin.api.DataSourceChannel;
 import org.apache.seatunnel.datasource.plugin.api.DataSourcePluginException;
 import org.apache.seatunnel.datasource.plugin.api.model.TableField;
@@ -41,8 +40,6 @@ public class StarRocksDataSourceChannel implements DataSourceChannel {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(StarRocksDataSourceChannel.class);
 
-    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
-
     @Override
     public boolean canAbleGetSchema() {
         return true;
@@ -81,7 +78,7 @@ public boolean checkDataSourceConnectivity(
         try {
             StarRocksCatalog catalog = getCatalog(requestParams);
             String nodeUrls = requestParams.get(StarRocksOptionRule.NODE_URLS.key());
-            List<String> nodeList = OBJECT_MAPPER.readValue(nodeUrls, List.class);
+            List<String> nodeList = JsonUtils.toList(nodeUrls, String.class);
             if (!telnet(nodeList.get(0))) {
                 return false;
             }

diff --git a/...-app/src/main/java/org/apache/seatunnel/app/controller/SeatunnelDatasourceController.java b/...-app/src/main/java/org/apache/seatunnel/app/controller/SeatunnelDatasourceController.java
@@ -31,8 +31,8 @@
 import org.apache.seatunnel.app.domain.response.datasource.DatasourceRes;
 import org.apache.seatunnel.app.service.IDatasourceService;
 import org.apache.seatunnel.app.utils.CartesianProductUtils;
-import org.apache.seatunnel.app.utils.JSONUtils;
 import org.apache.seatunnel.app.utils.PropertyUtils;
+import org.apache.seatunnel.common.utils.JsonUtils;
 import org.apache.seatunnel.datasource.plugin.api.DataSourcePluginInfo;
 import org.apache.seatunnel.datasource.plugin.api.model.TableField;
 import org.apache.seatunnel.server.common.SeatunnelErrorEnum;
@@ -116,7 +116,7 @@ Result<String> createDatasource(
             @ApiIgnore @RequestAttribute(value = SESSION_USER) User loginUser,
             @RequestBody DatasourceReq req) {
         String datasourceConfig = req.getDatasourceConfig();
-        Map<String, String> stringStringMap = JSONUtils.toMap(datasourceConfig);
+        Map<String, String> stringStringMap = JsonUtils.toMap(datasourceConfig);
         return Result.success(
                 datasourceService.createDatasource(
                         loginUser.getId(),
@@ -146,8 +146,6 @@ Result<String> createDatasource(
     Result<Boolean> testConnect(
             @ApiIgnore @RequestAttribute(value = SESSION_USER) User loginUser,
             @RequestBody DatasourceCheckReq req) {
-
-        // Map<String, String> stringStringMap = JSONUtils.toMap(req.getDatasourceConfig());
         return Result.success(
                 datasourceService.testDatasourceConnectionAble(
                         loginUser.getId(),
@@ -182,7 +180,7 @@ Result<Boolean> updateDatasource(
             @ApiIgnore @RequestAttribute(value = SESSION_USER) User loginUser,
             @PathVariable("id") String id,
             @RequestBody DatasourceReq req) {
-        Map<String, String> stringStringMap = JSONUtils.toMap(req.getDatasourceConfig());
+        Map<String, String> stringStringMap = JsonUtils.toMap(req.getDatasourceConfig());
         Long datasourceId = Long.parseLong(id);
         return Result.success(
                 datasourceService.updateDatasource(

diff --git a/.../seatunnel-app/src/main/java/org/apache/seatunnel/app/dal/entity/ProcessTaskRelation.java b/.../seatunnel-app/src/main/java/org/apache/seatunnel/app/dal/entity/ProcessTaskRelation.java
@@ -17,14 +17,15 @@
 
 package org.apache.seatunnel.app.dal.entity;
 
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
 import org.apache.seatunnel.app.common.ConditionType;
-import org.apache.seatunnel.app.utils.JSONUtils;
+import org.apache.seatunnel.common.utils.JsonUtils;
 
 import com.baomidou.mybatisplus.annotation.IdType;
 import com.baomidou.mybatisplus.annotation.TableId;
 import com.baomidou.mybatisplus.annotation.TableName;
-import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
-import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import lombok.Data;
 
 import java.util.Date;
@@ -66,8 +67,8 @@ public class ProcessTaskRelation {
     private ConditionType conditionType;
 
     /** condition parameters */
-    @JsonDeserialize(using = JSONUtils.JsonDataDeserializer.class)
-    @JsonSerialize(using = JSONUtils.JsonDataSerializer.class)
+    @JsonDeserialize(using = JsonUtils.JsonDataDeserializer.class)
+    @JsonSerialize(using = JsonUtils.JsonDataSerializer.class)
     private String conditionParams;
 
     /** create time */

diff --git a/...erver/seatunnel-app/src/main/java/org/apache/seatunnel/app/dal/entity/TaskDefinition.java b/...erver/seatunnel-app/src/main/java/org/apache/seatunnel/app/dal/entity/TaskDefinition.java
@@ -17,13 +17,18 @@
 
 package org.apache.seatunnel.app.dal.entity;
 
+import org.apache.seatunnel.shade.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
 import org.apache.seatunnel.app.common.Constants;
 import org.apache.seatunnel.app.common.Flag;
 import org.apache.seatunnel.app.common.Priority;
 import org.apache.seatunnel.app.common.TaskTimeoutStrategy;
 import org.apache.seatunnel.app.common.TimeoutFlag;
 import org.apache.seatunnel.app.domain.model.Property;
-import org.apache.seatunnel.app.utils.JSONUtils;
+import org.apache.seatunnel.common.utils.JsonUtils;
 
 import org.apache.commons.collections4.CollectionUtils;
 
@@ -32,9 +37,6 @@
 import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableId;
 import com.baomidou.mybatisplus.annotation.TableName;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
-import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import com.google.common.base.Strings;
 
 import java.util.Date;
@@ -73,8 +75,8 @@ public class TaskDefinition {
     private String taskType;
 
     /** user defined parameters */
-    @JsonDeserialize(using = JSONUtils.JsonDataDeserializer.class)
-    @JsonSerialize(using = JSONUtils.JsonDataSerializer.class)
+    @JsonDeserialize(using = JsonUtils.JsonDataDeserializer.class)
+    @JsonSerialize(using = JsonUtils.JsonDataSerializer.class)
     private String taskParams;
 
     /** user defined parameter list */
@@ -230,9 +232,9 @@ public void setTaskParams(String taskParams) {
     }
 
     public List<Property> getTaskParamList() {
-        JsonNode localParams = JSONUtils.parseObject(taskParams).findValue("localParams");
+        JsonNode localParams = JsonUtils.parseObject(taskParams).findValue("localParams");
         if (localParams != null) {
-            taskParamList = JSONUtils.toList(localParams.toString(), Property.class);
+            taskParamList = JsonUtils.toList(localParams.toString(), Property.class);
         }
 
         return taskParamList;
@@ -248,12 +250,12 @@ public void setTaskParamMap(Map<String, String> taskParamMap) {
 
     public Map<String, String> getTaskParamMap() {
         if (taskParamMap == null && !Strings.isNullOrEmpty(taskParams)) {
-            JsonNode localParams = JSONUtils.parseObject(taskParams).findValue("localParams");
+            JsonNode localParams = JsonUtils.parseObject(taskParams).findValue("localParams");
 
             // If a jsonNode is null, not only use !=null, but also it should use the isNull method
             // to be estimated.
             if (localParams != null && !localParams.isNull()) {
-                List<Property> propList = JSONUtils.toList(localParams.toString(), Property.class);
+                List<Property> propList = JsonUtils.toList(localParams.toString(), Property.class);
 
                 if (CollectionUtils.isNotEmpty(propList)) {
                     taskParamMap = new HashMap<>();
@@ -379,7 +381,12 @@ public void setDelayTime(int delayTime) {
     }
 
     public String getDependence() {
-        return JSONUtils.getNodeString(this.taskParams, Constants.DEPENDENCE);
+        try {
+            JsonNode jsonNode = JsonUtils.stringToJsonNode(this.taskParams);
+            return JsonUtils.findValue(jsonNode, Constants.DEPENDENCE);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
     }
 
     public String getModifyBy() {

diff --git a/...seatunnel-app/src/main/java/org/apache/seatunnel/app/dal/entity/TaskDefinitionExpand.java b/...seatunnel-app/src/main/java/org/apache/seatunnel/app/dal/entity/TaskDefinitionExpand.java
@@ -17,26 +17,26 @@
 
 package org.apache.seatunnel.app.dal.entity;
 
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode;
+
 import org.apache.seatunnel.app.parameters.DependentParameters;
 import org.apache.seatunnel.app.parameters.SubProcessParameters;
-import org.apache.seatunnel.app.utils.JSONUtils;
-
-import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.seatunnel.common.utils.JsonUtils;
 
 import static org.apache.seatunnel.app.common.Constants.CMD_PARAM_SUB_PROCESS_DEFINE_CODE;
 
 public class TaskDefinitionExpand extends TaskDefinition {
 
     public SubProcessParameters getSubProcessParameters() {
         String parameter = super.getTaskParams();
-        ObjectNode parameterJson = JSONUtils.parseObject(parameter);
+        ObjectNode parameterJson = JsonUtils.parseObject(parameter);
         if (parameterJson.get(CMD_PARAM_SUB_PROCESS_DEFINE_CODE) != null) {
-            return JSONUtils.parseObject(parameter, SubProcessParameters.class);
+            return JsonUtils.parseObject(parameter, SubProcessParameters.class);
         }
         return null;
     }
 
     public DependentParameters getDependentParameters() {
-        return JSONUtils.parseObject(super.getDependence(), DependentParameters.class);
+        return JsonUtils.parseObject(super.getDependence(), DependentParameters.class);
     }
 }
diff --git a/...r/seatunnel-app/src/main/java/org/apache/seatunnel/app/parameters/AbstractParameters.java b/...r/seatunnel-app/src/main/java/org/apache/seatunnel/app/parameters/AbstractParameters.java
@@ -17,18 +17,18 @@
 
 package org.apache.seatunnel.app.parameters;
 
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ArrayNode;
+
 import org.apache.seatunnel.app.common.Direct;
 import org.apache.seatunnel.app.domain.model.Property;
 import org.apache.seatunnel.app.domain.model.ResourceInfo;
 import org.apache.seatunnel.app.parameters.resource.ResourceParametersHelper;
-import org.apache.seatunnel.app.utils.JSONUtils;
+import org.apache.seatunnel.common.utils.JsonUtils;
 
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.node.ArrayNode;
-
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
@@ -121,7 +121,7 @@ public void setVarPool(String varPool) {
         if (StringUtils.isEmpty(varPool)) {
             this.varPool = new ArrayList<>();
         } else {
-            this.varPool = JSONUtils.toList(varPool, Property.class);
+            this.varPool = JsonUtils.toList(varPool, Property.class);
         }
     }
 
@@ -165,9 +165,9 @@ public List<Property> getOutProperty(List<Property> params) {
 
     public List<Map<String, String>> getListMapByString(String json) {
         List<Map<String, String>> allParams = new ArrayList<>();
-        ArrayNode paramsByJson = JSONUtils.parseArray(json);
+        ArrayNode paramsByJson = JsonUtils.parseArray(json);
         for (JsonNode jsonNode : paramsByJson) {
-            Map<String, String> param = JSONUtils.toMap(jsonNode.toString());
+            Map<String, String> param = JsonUtils.toMap(jsonNode.toString());
             allParams.add(param);
         }
         return allParams;

diff --git a/...nel-app/src/main/java/org/apache/seatunnel/app/parameters/resource/UdfFuncParameters.java b/...nel-app/src/main/java/org/apache/seatunnel/app/parameters/resource/UdfFuncParameters.java
@@ -18,7 +18,7 @@
 package org.apache.seatunnel.app.parameters.resource;
 
 import org.apache.seatunnel.app.common.UdfType;
-import org.apache.seatunnel.app.utils.JSONUtils;
+import org.apache.seatunnel.common.utils.JsonUtils;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 
@@ -218,6 +218,6 @@ public int hashCode() {
 
     @Override
     public String toString() {
-        return JSONUtils.toJsonString(this);
+        return JsonUtils.toJsonString(this);
     }
 }
diff --git a/...atunnel-app/src/main/java/org/apache/seatunnel/app/service/impl/JobConfigServiceImpl.java b/...atunnel-app/src/main/java/org/apache/seatunnel/app/service/impl/JobConfigServiceImpl.java
@@ -25,6 +25,7 @@
 import org.apache.seatunnel.app.permission.constants.SeatunnelFuncPermissionKeyConstant;
 import org.apache.seatunnel.app.service.IJobConfigService;
 import org.apache.seatunnel.common.constants.JobMode;
+import org.apache.seatunnel.common.utils.JsonUtils;
 import org.apache.seatunnel.server.common.SeatunnelErrorEnum;
 import org.apache.seatunnel.server.common.SeatunnelException;
 
@@ -34,18 +35,11 @@
 import org.springframework.transaction.annotation.Transactional;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
 
 import javax.annotation.Resource;
 
-import java.io.IOException;
-import java.util.Map;
-
 @Service
 public class JobConfigServiceImpl extends SeatunnelBaseServiceImpl implements IJobConfigService {
-
-    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
-
     private static final String JOB_MODE = "job.mode";
 
     @Resource private IJobVersionDao jobVersionDao;
@@ -65,14 +59,10 @@ public JobConfigRes getJobConfig(long jobVersionId) throws JsonProcessingExcepti
         jobConfigRes.setName(jobDefinition.getName());
         jobConfigRes.setId(jobVersion.getId());
         jobConfigRes.setDescription(jobDefinition.getDescription());
-        try {
-            jobConfigRes.setEnv(
-                    StringUtils.isEmpty(jobVersion.getEnv())
-                            ? null
-                            : OBJECT_MAPPER.readValue(jobVersion.getEnv(), Map.class));
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        }
+        jobConfigRes.setEnv(
+                StringUtils.isEmpty(jobVersion.getEnv())
+                        ? null
+                        : JsonUtils.toMap(jobVersion.getEnv(), String.class, Object.class));
         jobConfigRes.setEngine(jobVersion.getEngineName());
         return jobConfigRes;
     }
@@ -102,7 +92,7 @@ public void updateJobConfig(int userId, long jobVersionId, JobConfig jobConfig)
                             .jobMode(jobMode)
                             .engineName(jobConfig.getEngine())
                             .updateUserId(userId)
-                            .env(OBJECT_MAPPER.writeValueAsString(jobConfig.getEnv()))
+                            .env(JsonUtils.toJsonString(jobConfig.getEnv()))
                             .build());
         } else {
             throw new SeatunnelException(SeatunnelErrorEnum.ILLEGAL_STATE, "job mode is not set");