Repository: NetEase/amoro (java)

Issue:
[Improvement]: Replace fastjson to jackson
### Search before asking

- [X] I have searched in the [issues](https://github.com/NetEase/arctic/issues?q=is%3Aissue) and found no similar issues.


### What would you like to be improved?

replace fastjson to jackson

### How should we improve?

Fastjson has too many vulnerabilities, maybe we should replace it.

In the last few years, fastjson has been exposed to multiple vulnerabilities. It has even been exposed to CRITICAL-level security vulnerabilities,such as CVE-2017-18349 and other vulnerabilities.
In order to ensure the security of the project, I think fastjson should be replaced and jackson should be used.

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!

### Subtasks

_No response_

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)


Pull request description:
[AMORO-1123] Replace fastjson with jackson
<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://amoro.netease.com/how-to-contribute/
  2. If the PR is related to an issue in https://github.com/NetEase/amoro/issues, add '[AMORO-XXXX]' in your PR title, e.g., '[AMORO-XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][AMORO-XXXX] Your PR title ...'.
-->

## Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
  1. If you add a feature, you can talk about its use case.
  2. If you fix a bug, you can clarify why it is a bug.
  3. Use Fix/Resolve/Close #{ISSUE_NUMBER} to link this PR to its related issue
-->

Close #1123 .

## Brief change log
<!--
Clearly describe the changes made in modules, classes, methods, etc.
-->

- Replace fastjson with jackson

## How was this patch tested?

- [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [ ] Run test locally before making a pull request

## Documentation

- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)


Patch:
diff --git a/ams/server/src/main/java/com/netease/arctic/server/ArcticServiceContainer.java b/ams/server/src/main/java/com/netease/arctic/server/ArcticServiceContainer.java
index 9d42628b1..bc7822ae9 100644
--- a/ams/server/src/main/java/com/netease/arctic/server/ArcticServiceContainer.java
+++ b/ams/server/src/main/java/com/netease/arctic/server/ArcticServiceContainer.java
@@ -18,8 +18,6 @@
 
 package com.netease.arctic.server;
 
-import com.alibaba.fastjson.JSONArray;
-import com.alibaba.fastjson.JSONObject;
 import com.google.common.collect.Maps;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.netease.arctic.Constants;
@@ -46,11 +44,14 @@ import com.netease.arctic.server.utils.ConfigOption;
 import com.netease.arctic.server.utils.ConfigurationUtil;
 import com.netease.arctic.server.utils.Configurations;
 import com.netease.arctic.server.utils.ThriftServiceProxy;
+import com.netease.arctic.utils.JacksonUtil;
 import io.javalin.Javalin;
 import io.javalin.http.HttpCode;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.iceberg.SystemProperties;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
 import org.apache.thrift.TMultiplexedProcessor;
 import org.apache.thrift.TProcessor;
 import org.apache.thrift.protocol.TBinaryProtocol;
@@ -377,7 +378,7 @@ public class ArcticServiceContainer {
 
   private class ConfigurationHelper {
 
-    private JSONObject yamlConfig;
+    private JsonNode yamlConfig;
 
     public void init() throws IOException {
       Map<String, Object> envConfig = initEnvConfig();
@@ -386,14 +387,18 @@ public class ArcticServiceContainer {
       initContainerConfig();
     }
 
-    @SuppressWarnings("unchecked")
     private void initServiceConfig(Map<String, Object> envConfig) throws IOException {
       LOG.info("initializing service configuration...");
       String configPath = Environments.getConfigPath() + "/" + SERVER_CONFIG_FILENAME;
       LOG.info("load config from path: {}", configPath);
       yamlConfig =
-          new JSONObject(new Yaml().loadAs(Files.newInputStream(Paths.get(configPath)), Map.class));
-      JSONObject systemConfig = yamlConfig.getJSONObject(ArcticManagementConf.SYSTEM_CONFIG);
+          JacksonUtil.fromObjects(
+              new Yaml().loadAs(Files.newInputStream(Paths.get(configPath)), Map.class));
+      Map<String, Object> systemConfig =
+          JacksonUtil.getMap(
+              yamlConfig,
+              ArcticManagementConf.SYSTEM_CONFIG,
+              new TypeReference<Map<String, Object>>() {});
       Map<String, Object> expandedConfigurationMap = Maps.newHashMap();
       expandConfigMap(systemConfig, "", expandedConfigurationMap);
       // If same configurations in files and environment variables, environment variables have
@@ -493,23 +498,24 @@ public class ArcticServiceContainer {
           SystemProperties.WORKER_THREAD_POOL_SIZE_PROP, String.valueOf(workerThreadPoolSize));
     }
 
-    @SuppressWarnings("unchecked")
     private void initContainerConfig() {
       LOG.info("initializing container configuration...");
-      JSONArray containers = yamlConfig.getJSONArray(ArcticManagementConf.CONTAINER_LIST);
+      JsonNode containers = yamlConfig.get(ArcticManagementConf.CONTAINER_LIST);
       List<ContainerMetadata> containerList = new ArrayList<>();
-      if (containers != null) {
-        for (int i = 0; i < containers.size(); i++) {
-          JSONObject containerConfig = containers.getJSONObject(i);
+      if (containers != null && containers.isArray()) {
+        for (final JsonNode containerConfig : containers) {
           ContainerMetadata container =
               new ContainerMetadata(
-                  containerConfig.getString(ArcticManagementConf.CONTAINER_NAME),
-                  containerConfig.getString(ArcticManagementConf.CONTAINER_IMPL));
-          Map<String, String> containerProperties = new HashMap<>();
-          if (containerConfig.containsKey(ArcticManagementConf.CONTAINER_PROPERTIES)) {
-            containerProperties.putAll(
-                containerConfig.getObject(ArcticManagementConf.CONTAINER_PROPERTIES, Map.class));
-          }
+                  containerConfig.get(ArcticManagementConf.CONTAINER_NAME).asText(),
+                  containerConfig.get(ArcticManagementConf.CONTAINER_IMPL).asText());
+
+          Map<String, String> containerProperties =
+              new HashMap<>(
+                  JacksonUtil.getMap(
+                      containerConfig,
+                      ArcticManagementConf.CONTAINER_PROPERTIES,
+                      new TypeReference<Map<String, String>>() {}));
+
           // put properties in config.yaml first.
           containerProperties.put(OptimizerProperties.AMS_HOME, Environments.getHomePath());
           containerProperties.putIfAbsent(
diff --git a/ams/server/src/main/java/com/netease/arctic/server/HighAvailabilityContainer.java b/ams/server/src/main/java/com/netease/arctic/server/HighAvailabilityContainer.java
index 66a0a8d2a..9ca987492 100644
--- a/ams/server/src/main/java/com/netease/arctic/server/HighAvailabilityContainer.java
+++ b/ams/server/src/main/java/com/netease/arctic/server/HighAvailabilityContainer.java
@@ -18,10 +18,10 @@
 
 package com.netease.arctic.server;
 
-import com.alibaba.fastjson.JSONObject;
 import com.netease.arctic.client.AmsServerInfo;
 import com.netease.arctic.properties.AmsHAProperties;
 import com.netease.arctic.server.utils.Configurations;
+import com.netease.arctic.utils.JacksonUtil;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.framework.recipes.leader.LeaderLatch;
@@ -99,12 +99,12 @@ public class HighAvailabilityContainer implements LeaderLatchListener {
             .setData()
             .forPath(
                 tableServiceMasterPath,
-                JSONObject.toJSONString(tableServiceServerInfo).getBytes(StandardCharsets.UTF_8));
+                JacksonUtil.toJSONString(tableServiceServerInfo).getBytes(StandardCharsets.UTF_8));
         zkClient
             .setData()
             .forPath(
                 optimizingServiceMasterPath,
-                JSONObject.toJSONString(optimizingServiceServerInfo)
+                JacksonUtil.toJSONString(optimizingServiceServerInfo)
                     .getBytes(StandardCharsets.UTF_8));
       }
     }
diff --git a/ams/server/src/main/java/com/netease/arctic/server/dashboard/DashboardServer.java b/ams/server/src/main/java/com/netease/arctic/server/dashboard/DashboardServer.java
index ee7df7abe..af01730d4 100644
--- a/ams/server/src/main/java/com/netease/arctic/server/dashboard/DashboardServer.java
+++ b/ams/server/src/main/java/com/netease/arctic/server/dashboard/DashboardServer.java
@@ -24,7 +24,6 @@ import static io.javalin.apibuilder.ApiBuilder.path;
 import static io.javalin.apibuilder.ApiBuilder.post;
 import static io.javalin.apibuilder.ApiBuilder.put;
 
-import com.alibaba.fastjson.JSONObject;
 import com.netease.arctic.server.DefaultOptimizingService;
 import com.netease.arctic.server.RestCatalogService;
 import com.netease.arctic.server.dashboard.controller.CatalogController;
@@ -43,6 +42,7 @@ import com.netease.arctic.server.exception.SignatureCheckException;
 import com.netease.arctic.server.table.TableService;
 import com.netease.arctic.server.terminal.TerminalManager;
 import com.netease.arctic.server.utils.Configurations;
+import com.netease.arctic.utils.JacksonUtil;
 import io.javalin.apibuilder.EndpointGroup;
 import io.javalin.http.ContentType;
 import io.javalin.http.Context;
@@ -396,7 +396,7 @@ public class DashboardServer {
     } else if (needLoginCheck(uriPath)) {
       if (null == ctx.sessionAttribute("user")) {
         ctx.sessionAttributeMap();
-        LOG.info("session info: {}", JSONObject.toJSONString(ctx.sessionAttributeMap()));
+        LOG.info("session info: {}", JacksonUtil.toJSONString(ctx.sessionAttributeMap()));
         throw new ForbiddenException();
       }
     }
diff --git a/ams/server/src/main/java/com/netease/arctic/server/dashboard/controller/LoginController.java b/ams/server/src/main/java/com/netease/arctic/server/dashboard/controller/LoginController.java
index d77b359c0..32acf84e3 100644
--- a/ams/server/src/main/java/com/netease/arctic/server/dashboard/controller/LoginController.java
+++ b/ams/server/src/main/java/com/netease/arctic/server/dashboard/controller/LoginController.java
@@ -18,11 +18,12 @@
 
 package com.netease.arctic.server.dashboard.controller;
 
-import com.alibaba.fastjson.JSONObject;
 import com.netease.arctic.server.ArcticManagementConf;
 import com.netease.arctic.server.dashboard.response.OkResponse;
 import com.netease.arctic.server.utils.Configurations;
+import com.netease.arctic.utils.JacksonUtil;
 import io.javalin.http.Context;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
 
 import java.io.Serializable;
 
@@ -46,13 +47,14 @@ public class LoginController {
   /** handle login post request. */
   public void login(Context ctx) {
     // ok
-    JSONObject postBody = ctx.bodyAsClass(JSONObject.class);
-    if (adminUser.equals(postBody.get("user"))
-        && (adminPassword.equals(postBody.get("password")))) {
+    JsonNode postBody = ctx.bodyAsClass(JsonNode.class);
+    String user = JacksonUtil.getString(postBody, "user");
+    String pwd = JacksonUtil.getString(postBody, "password");
+    if (adminUser.equals(user) && (adminPassword.equals(pwd))) {
       ctx.sessionAttribute("user", new SessionInfo(adminUser, System.currentTimeMillis() + ""));
       ctx.json(OkResponse.of("success"));
     } else {
-      throw new RuntimeException("bad user " + postBody.get("user") + " or password!");
+      throw new RuntimeException("bad user " + user + " or password!");
     }
   }
 
diff --git a/ams/server/src/main/java/com/netease/arctic/server/dashboard/controller/OptimizerController.java b/ams/server/src/main/java/com/netease/arctic/server/dashboard/controller/OptimizerController.java
index e7bcf5e91..97ba2dd10 100644
--- a/ams/server/src/main/java/com/netease/arctic/server/dashboard/controller/OptimizerController.java
+++ b/ams/server/src/main/java/com/netease/arctic/server/dashboard/controller/OptimizerController.java
@@ -18,8 +18,6 @@
 
 package com.netease.arctic.server.dashboard.controller;
 
-import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.JSONObject;
 import com.google.common.base.Preconditions;
 import com.netease.arctic.api.resource.Resource;
 import com.netease.arctic.api.resource.ResourceGroup;
@@ -36,7 +34,10 @@ import com.netease.arctic.server.resource.ResourceContainers;
 import com.netease.arctic.server.table.ServerTableIdentifier;
 import com.netease.arctic.server.table.TableRuntime;
 import com.netease.arctic.server.table.TableService;
+import com.netease.arctic.utils.JacksonUtil;
 import io.javalin.http.Context;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
 
 import javax.ws.rs.BadRequestException;
 
@@ -109,11 +110,11 @@ public class OptimizerController {
     }
     List<OptimizerInstance> optimizerList = new ArrayList<>(optimizers);
     optimizerList.sort(Comparator.comparingLong(OptimizerInstance::getStartTime).reversed());
-    List<JSONObject> result =
+    List<JsonNode> result =
         optimizerList.stream()
             .map(
                 e -> {
-                  JSONObject jsonObject = (JSONObject) JSON.toJSON(e);
+                  ObjectNode jsonObject = (ObjectNode) JacksonUtil.fromObjects(e);
                   jsonObject.put("jobId", e.getResourceId());
                   jsonObject.put("optimizerGroup", e.getGroupName());
                   jsonObject.put("coreNumber", e.getThreadCount());
@@ -124,13 +125,13 @@ public class OptimizerController {
                 })
             .collect(Collectors.toList());
 
-    PageResult<JSONObject> amsPageResult = PageResult.of(result, offset, pageSize);
+    PageResult<JsonNode> amsPageResult = PageResult.of(result, offset, pageSize);
     ctx.json(OkResponse.of(amsPageResult));
   }
 
   /** get optimizerGroup: optimizerGroupId, optimizerGroupName url = /optimizerGroups. */
   public void getOptimizerGroups(Context ctx) {
-    List<JSONObject> result =
+    List<JsonNode> result =
         optimizerManager.listResourceGroups().stream()
             .filter(
                 resourceGroup ->
@@ -138,7 +139,8 @@ public class OptimizerController {
                         resourceGroup.getContainer()))
             .map(
                 e -> {
-                  JSONObject jsonObject = new JSONObject();
+                  ObjectNode jsonObject = JacksonUtil.createEmptyObjectNode();
+
                   jsonObject.put("optimizerGroupName", e.getName());
                   return jsonObject;
                 })
diff --git a/ams/server/src/main/java/com/netease/arctic/server/manager/AbstractPluginManager.java b/ams/server/src/main/java/com/netease/arctic/server/manager/AbstractPluginManager.java
index a644f3c2b..39e792def 100644
--- a/ams/server/src/main/java/com/netease/arctic/server/manager/AbstractPluginManager.java
+++ b/ams/server/src/main/java/com/netease/arctic/server/manager/AbstractPluginManager.java
@@ -18,17 +18,17 @@
 
 package com.netease.arctic.server.manager;
 
-import com.alibaba.fastjson.JSONArray;
-import com.alibaba.fastjson.JSONObject;
 import com.netease.arctic.api.ActivePlugin;
 import com.netease.arctic.server.Environments;
 import com.netease.arctic.server.exception.AlreadyExistsException;
 import com.netease.arctic.server.exception.LoadingPluginException;
+import com.netease.arctic.utils.JacksonUtil;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.yaml.snakeyaml.Yaml;
@@ -43,7 +43,12 @@ import java.net.URLClassLoader;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.ServiceLoader;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
@@ -240,7 +245,7 @@ public abstract class AbstractPluginManager<T extends ActivePlugin> implements P
 
   @VisibleForTesting
   protected List<PluginConfiguration> loadPluginConfigurations() {
-    JSONObject yamlConfig = null;
+    JsonNode yamlConfig = null;
     Path mangerConfigPath = pluginManagerConfigFilePath();
     if (!Files.exists(mangerConfigPath)) {
       return ImmutableList.of();
@@ -248,7 +253,7 @@ public abstract class AbstractPluginManager<T extends ActivePlugin> implements P
     try {
       Object yamlObj = new Yaml().loadAs(Files.newInputStream(mangerConfigPath), Object.class);
       if (yamlObj instanceof Map) {
-        yamlConfig = new JSONObject((Map) yamlObj);
+        yamlConfig = JacksonUtil.fromObjects(yamlObj);
       }
     } catch (IOException e) {
       throw new LoadingPluginException(
@@ -258,11 +263,11 @@ public abstract class AbstractPluginManager<T extends ActivePlugin> implements P
     LOG.info("initializing plugin configuration for: " + pluginCategory());
     String pluginListKey = pluginCategory();
 
-    JSONArray pluginConfigList = yamlConfig != null ? yamlConfig.getJSONArray(pluginListKey) : null;
+    JsonNode pluginConfigList = yamlConfig != null ? yamlConfig.get(pluginListKey) : null;
     List<PluginConfiguration> configs = Lists.newArrayList();
     if (pluginConfigList != null && !pluginConfigList.isEmpty()) {
       for (int i = 0; i < pluginConfigList.size(); i++) {
-        JSONObject pluginConfiguration = pluginConfigList.getJSONObject(i);
+        JsonNode pluginConfiguration = pluginConfigList.get(i);
         PluginConfiguration configuration = PluginConfiguration.fromJSONObject(pluginConfiguration);
         configs.add(configuration);
       }
diff --git a/ams/server/src/main/java/com/netease/arctic/server/manager/PluginConfiguration.java b/ams/server/src/main/java/com/netease/arctic/server/manager/PluginConfiguration.java
index 86686cad4..5134588cf 100644
--- a/ams/server/src/main/java/com/netease/arctic/server/manager/PluginConfiguration.java
+++ b/ams/server/src/main/java/com/netease/arctic/server/manager/PluginConfiguration.java
@@ -18,14 +18,14 @@
 
 package com.netease.arctic.server.manager;
 
-import com.alibaba.fastjson.JSONObject;
-import com.alibaba.fastjson.TypeReference;
+import com.netease.arctic.utils.JacksonUtil;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
 
 import java.util.Collections;
 import java.util.Map;
-import java.util.Optional;
 
 /** Configuration of a plugin. */
 public class PluginConfiguration {
@@ -42,12 +42,14 @@ public class PluginConfiguration {
     this.properties = properties;
   }
 
-  public static PluginConfiguration fromJSONObject(JSONObject configOptions) {
-    Preconditions.checkArgument(configOptions.containsKey(NAME), "plugin name is required");
-    String name = configOptions.get(NAME).toString();
-    boolean enabled = Optional.ofNullable(configOptions.getBoolean(ENABLED)).orElse(true);
+  public static PluginConfiguration fromJSONObject(JsonNode configOptions) {
+    JsonNode nameNode = configOptions.get(NAME);
+    Preconditions.checkNotNull(nameNode, "plugin name is required");
+    String name = nameNode.textValue();
+
+    boolean enabled = JacksonUtil.getBoolean(configOptions, ENABLED, true);
     Map<String, String> props =
-        configOptions.getObject(PROPERTIES, new TypeReference<Map<String, String>>() {});
+        JacksonUtil.getMap(configOptions, PROPERTIES, new TypeReference<Map<String, String>>() {});
     if (props == null) {
       props = ImmutableMap.of();
     }
diff --git a/core/pom.xml b/core/pom.xml
index 7933fa575..ef98bf57a 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -59,6 +59,11 @@
             <artifactId>curator-recipes</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.paimon</groupId>
+            <artifactId>paimon-shade-jackson-2</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.zookeeper</groupId>
             <artifactId>zookeeper</artifactId>
@@ -80,11 +85,6 @@
             <artifactId>commons-pool2</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>com.alibaba</groupId>
-            <artifactId>fastjson</artifactId>
-        </dependency>
-
         <dependency>
             <groupId>org.rocksdb</groupId>
             <artifactId>rocksdbjni</artifactId>
diff --git a/core/src/main/java/com/netease/arctic/client/ArcticThriftUrl.java b/core/src/main/java/com/netease/arctic/client/ArcticThriftUrl.java
index 38bea8f70..dd3e6262a 100644
--- a/core/src/main/java/com/netease/arctic/client/ArcticThriftUrl.java
+++ b/core/src/main/java/com/netease/arctic/client/ArcticThriftUrl.java
@@ -18,9 +18,9 @@
 
 package com.netease.arctic.client;
 
-import com.alibaba.fastjson.JSONObject;
 import com.netease.arctic.Constants;
 import com.netease.arctic.properties.AmsHAProperties;
+import com.netease.arctic.utils.JacksonUtil;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.zookeeper.KeeperException;
 import org.slf4j.Logger;
@@ -194,12 +194,12 @@ public class ArcticThriftUrl {
       String serviceName, String zkServerAddress, String cluster) throws Exception {
     switch (serviceName) {
       case Constants.THRIFT_TABLE_SERVICE_NAME:
-        return JSONObject.parseObject(
+        return JacksonUtil.parseObject(
             ZookeeperService.getInstance(zkServerAddress)
                 .getData(AmsHAProperties.getTableServiceMasterPath(cluster)),
             AmsServerInfo.class);
       case Constants.THRIFT_OPTIMIZING_SERVICE_NAME:
-        return JSONObject.parseObject(
+        return JacksonUtil.parseObject(
             ZookeeperService.getInstance(zkServerAddress)
                 .getData(AmsHAProperties.getOptimizingServiceMasterPath(cluster)),
             AmsServerInfo.class);
diff --git a/core/src/main/java/com/netease/arctic/utils/JacksonUtil.java b/core/src/main/java/com/netease/arctic/utils/JacksonUtil.java
new file mode 100644
index 000000000..5b4fa1261
--- /dev/null
+++ b/core/src/main/java/com/netease/arctic/utils/JacksonUtil.java
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netease.arctic.utils;
+
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.DeserializationFeature;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.SerializationFeature;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+/** Util class for Jackson. */
+public class JacksonUtil {
+
+  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+  private static final ObjectMapper INDENT_OBJECT_MAPPER = new ObjectMapper();
+
+  private static final String SERIALIZE_ERROR = "serialize object error";
+
+  private static final String DESERIALIZE_ERROR = "deserialize to object error";
+
+  static {
+    OBJECT_MAPPER.configure(SerializationFeature.WRITE_ENUMS_USING_INDEX, true);
+    OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+    INDENT_OBJECT_MAPPER.configure(SerializationFeature.INDENT_OUTPUT, true);
+  }
+
+  private JacksonUtil() {
+    // do nothing
+  }
+
+  public static ObjectNode createEmptyObjectNode() {
+    return OBJECT_MAPPER.createObjectNode();
+  }
+
+  public static String toJSONString(Object object) {
+    try {
+      return OBJECT_MAPPER.writeValueAsString(object);
+    } catch (JsonProcessingException e) {
+      throw new IllegalStateException(SERIALIZE_ERROR, e);
+    }
+  }
+
+  /**
+   * Deserialize a json object from string which serialized using {@link #toJSONString(Object)} }.
+   */
+  public static <T> T parseObject(String jsonString, Class<T> clazz) {
+    try {
+      return OBJECT_MAPPER.readValue(jsonString, clazz);
+    } catch (IOException e) {
+      throw new IllegalStateException(DESERIALIZE_ERROR, e);
+    }
+  }
+
+  /** Convert an java object(usually java bean) to JsonNode. */
+  public static <T> JsonNode fromObjects(T object) {
+    return OBJECT_MAPPER.valueToTree(object);
+  }
+
+  public static Integer getInteger(JsonNode jsonNode, String fieldName) {
+    Preconditions.checkNotNull(jsonNode, "JsonNode should not be null.");
+
+    JsonNode valueNode = jsonNode.get(fieldName);
+    if (valueNode == null) {
+      return null;
+    }
+
+    if (valueNode.isInt()) {
+      return valueNode.intValue();
+    }
+    throw new ClassCastException("can not cast to int, value : " + valueNode);
+  }
+
+  public static Boolean getBoolean(JsonNode jsonNode, String fieldName, boolean defaultValue) {
+    Preconditions.checkNotNull(jsonNode, "JsonNode should not be null.");
+    JsonNode value = jsonNode.get(fieldName);
+    if (value == null) {
+      return defaultValue;
+    }
+
+    if (value.isBoolean()) {
+      return value.booleanValue();
+    }
+    throw new ClassCastException("can not cast to bool, value : " + value);
+  }
+
+  public static Boolean getBoolean(JsonNode jsonNode, String fieldName) {
+    return getBoolean(jsonNode, fieldName, false);
+  }
+
+  public static String getString(JsonNode jsonNode, String fieldName) {
+    return getString(jsonNode, fieldName, null);
+  }
+
+  public static String getString(JsonNode jsonNode, String fieldName, String defaultValue) {
+    Preconditions.checkNotNull(jsonNode, "JsonNode should not be null");
+    JsonNode textNode = jsonNode.get(fieldName);
+
+    if (textNode == null) {
+      return defaultValue;
+    }
+    return textNode.asText();
+  }
+
+  /** Retrieve the given map from a JsonNode, and return empty map if the filed doesn't exist. */
+  public static <K, V> Map<K, V> getMap(
+      JsonNode jsonNode, String fieldName, TypeReference<Map<K, V>> mapType) {
+    Preconditions.checkNotNull(jsonNode, "JsonNode should not be null");
+
+    JsonNode mapNode = jsonNode.get(fieldName);
+    if (mapNode == null) {
+      return new HashMap<>();
+    }
+
+    return OBJECT_MAPPER.convertValue(mapNode, mapType);
+  }
+}
diff --git a/mixed-format/spark/v3.2/spark-runtime/pom.xml b/mixed-format/spark/v3.2/spark-runtime/pom.xml
index d1fe9d8b4..95c3fc029 100644
--- a/mixed-format/spark/v3.2/spark-runtime/pom.xml
+++ b/mixed-format/spark/v3.2/spark-runtime/pom.xml
@@ -63,7 +63,7 @@
                                     <include>com.netease.amoro:amoro-core</include>
                                     <include>com.netease.amoro:amoro-ams-api</include>
                                     <include>com.netease.amoro:amoro-mixed-format-hive</include>
-                                    <include>com.alibaba:fastjson</include>
+                                    <include>org.apache.paimon:paimon-shade-jackson-2</include>
                                     <include>org.apache.curator:curator-framework</include>
                                     <include>org.apache.curator:curator-client</include>
                                     <include>org.apache.curator:curator-recipes</include>
@@ -148,11 +148,6 @@
                                         </include>
                                     </includes>
                                 </relocation>
-                                <relocation>
-                                    <pattern>com.alibaba.fastjson</pattern>
-                                    <shadedPattern>com.netease.arctic.shade.com.alibaba.fastjson
-                                    </shadedPattern>
-                                </relocation>
 
                                 <relocation>
                                     <pattern>org.apache.zookeeper</pattern>
diff --git a/mixed-format/spark/v3.3/spark-runtime/pom.xml b/mixed-format/spark/v3.3/spark-runtime/pom.xml
index 686a45763..ea0272cc1 100644
--- a/mixed-format/spark/v3.3/spark-runtime/pom.xml
+++ b/mixed-format/spark/v3.3/spark-runtime/pom.xml
@@ -63,7 +63,7 @@
                                     <include>com.netease.amoro:amoro-core</include>
                                     <include>com.netease.amoro:amoro-ams-api</include>
                                     <include>com.netease.amoro:amoro-mixed-format-hive</include>
-                                    <include>com.alibaba:fastjson</include>
+                                    <include>org.apache.paimon:paimon-shade-jackson-2</include>
                                     <include>org.apache.curator:curator-framework</include>
                                     <include>org.apache.curator:curator-client</include>
                                     <include>org.apache.curator:curator-recipes</include>
@@ -148,11 +148,6 @@
                                         </include>
                                     </includes>
                                 </relocation>
-                                <relocation>
-                                    <pattern>com.alibaba.fastjson</pattern>
-                                    <shadedPattern>com.netease.arctic.shade.com.alibaba.fastjson
-                                    </shadedPattern>
-                                </relocation>
 
                                 <relocation>
                                     <pattern>org.apache.zookeeper</pattern>
diff --git a/pom.xml b/pom.xml
index ed440f942..2a1ab266d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -103,7 +103,7 @@
         <mockito.version>4.11.0</mockito.version>
         <curator.version>5.6.0</curator.version>
         <zookeeper.version>3.9.1</zookeeper.version>
-        <fastjson.version>1.2.83</fastjson.version>
+        <jackson.shaded.version>2.14.2-0.4.0-incubating</jackson.shaded.version>
         <parquet-avro.version>1.13.1</parquet-avro.version>
         <orc-core.version>1.8.3</orc-core.version>
         <awssdk.version>2.24.12</awssdk.version>
@@ -529,9 +529,9 @@
             </dependency>
 
             <dependency>
-                <groupId>com.alibaba</groupId>
-                <artifactId>fastjson</artifactId>
-                <version>${fastjson.version}</version>
+                <groupId>org.apache.paimon</groupId>
+                <artifactId>paimon-shade-jackson-2</artifactId>
+                <version>${jackson.shaded.version}</version>
             </dependency>
 
             <dependency>


Test patch:
diff --git a/core/src/test/java/com/netease/arctic/utils/JacksonUtilTest.java b/core/src/test/java/com/netease/arctic/utils/JacksonUtilTest.java
new file mode 100644
index 000000000..e13267aa8
--- /dev/null
+++ b/core/src/test/java/com/netease/arctic/utils/JacksonUtilTest.java
@@ -0,0 +1,195 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netease.arctic.utils;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+public class JacksonUtilTest {
+
+  @Test
+  public void jacksonTest() {
+
+    JsonTestBean testObject = new JsonTestBean();
+    testObject.setBoolValue(true);
+    testObject.setIntValue(3);
+    testObject.setStringValue("abcd");
+    Map<String, String> map = new HashMap<>();
+    map.put("a", "b");
+    testObject.setMapValue(map);
+
+    String jasonString = JacksonUtil.toJSONString(testObject);
+
+    String expectedValue =
+        "{\"stringValue\":\"abcd\",\"intValue\":3,\"boolValue\":true,\"mapValue\":{\"a\":\"b\"}}";
+    assertEquals(expectedValue, jasonString);
+
+    JsonTestBean objReadFromJson = JacksonUtil.parseObject(jasonString, JsonTestBean.class);
+    assertEquals(testObject, objReadFromJson);
+
+    JsonNode jsonNode = JacksonUtil.fromObjects(objReadFromJson);
+
+    Integer expectedInt = 3;
+    assertEquals(expectedInt, JacksonUtil.getInteger(jsonNode, "intValue"));
+    // return null if key does not exist
+    assertNull(JacksonUtil.getInteger(jsonNode, "NoIntValue"));
+
+    String expectedString = "abcd";
+    assertEquals(expectedString, JacksonUtil.getString(jsonNode, "stringValue"));
+    // return null if key does not exist
+    assertNull(JacksonUtil.getString(jsonNode, "NoStringValue"));
+    // return default value if key does not exist, and default value passed
+    assertEquals("DefaultValue", JacksonUtil.getString(jsonNode, "NoStringValue", "DefaultValue"));
+
+    assertTrue(JacksonUtil.getBoolean(jsonNode, "boolValue"));
+    // returns false if key does not exist
+    assertFalse(JacksonUtil.getBoolean(jsonNode, "NboolValue"));
+    assertTrue(JacksonUtil.getBoolean(jsonNode, "NboolValue", true));
+
+    Map<String, Object> yamlMap = new HashMap<>();
+    Map<String, Object> amsNode = new HashMap<>();
+    amsNode.put("admin-username", "admin");
+    amsNode.put("admin-password", "admin");
+    amsNode.put("server-bind-host", "0.0.0.0");
+    amsNode.put("server-expose-host", "127.0.0.1");
+
+    yamlMap.put("ams", amsNode);
+
+    Map<String, Object> expireSnapshots = new HashMap<>();
+    expireSnapshots.put("enabled", true);
+    expireSnapshots.put("thread-count", 10);
+    yamlMap.put("expire-snapshots", expireSnapshots);
+
+    List<Map<String, Object>> containersList = new ArrayList<>();
+    Map<String, Object> localContainer = new HashMap<>();
+    localContainer.put("name", "localContainer");
+    localContainer.put("type", "local");
+    localContainer.put("properties", Collections.singletonMap("hadoop_home", "/opt/hadoop"));
+
+    containersList.add(localContainer);
+
+    Map<String, Object> flinkContainer = new HashMap<>();
+    flinkContainer.put("name", "flinkContainer");
+    flinkContainer.put("type", "flink");
+    Map<String, String> pro = new HashMap<>();
+    pro.put("FLINK_HOME", "/opt/flink/");
+    pro.put("HADOOP_CONF_DIR", "/etc/hadoop/conf/");
+    pro.put("HADOOP_USER_NAME", "hadoop");
+    pro.put("JVM_ARGS", "-Djava.security.krb5.conf=/opt/krb5.conf");
+    pro.put("FLINK_CONF_DIR", "/etc/hadoop/conf/");
+    flinkContainer.put("properties", pro);
+    containersList.add(flinkContainer);
+
+    yamlMap.put("containers", containersList);
+
+    JsonNode yamlNode = JacksonUtil.fromObjects(yamlMap);
+
+    JsonNode expiredSnapshots = yamlNode.get("expire-snapshots");
+    assertTrue(JacksonUtil.getBoolean(expiredSnapshots, "enabled"));
+    Integer expectedThreadCount = 10;
+    assertEquals(expectedThreadCount, JacksonUtil.getInteger(expiredSnapshots, "thread-count"));
+    JsonNode containersNode = yamlNode.get("containers");
+    assertEquals(2, containersNode.size());
+    JsonNode secondNode = containersNode.get(1);
+    assertEquals("flink", JacksonUtil.getString(secondNode, "type"));
+    assertEquals("flinkContainer", JacksonUtil.getString(secondNode, "name"));
+    JsonNode propertyiesNode = secondNode.get("properties");
+    assertEquals(5, propertyiesNode.size());
+
+    assertEquals("/opt/flink/", JacksonUtil.getString(propertyiesNode, "FLINK_HOME"));
+    assertEquals("/etc/hadoop/conf/", JacksonUtil.getString(propertyiesNode, "HADOOP_CONF_DIR"));
+    assertEquals("hadoop", JacksonUtil.getString(propertyiesNode, "HADOOP_USER_NAME"));
+    assertEquals(
+        "-Djava.security.krb5.conf=/opt/krb5.conf",
+        JacksonUtil.getString(propertyiesNode, "JVM_ARGS"));
+    assertEquals("/etc/hadoop/conf/", JacksonUtil.getString(propertyiesNode, "FLINK_CONF_DIR"));
+  }
+
+  static class JsonTestBean {
+
+    private String stringValue;
+
+    private Integer intValue;
+
+    private Boolean boolValue;
+
+    private Map<String, String> mapValue;
+
+    public String getStringValue() {
+      return stringValue;
+    }
+
+    public void setStringValue(String stringValue) {
+      this.stringValue = stringValue;
+    }
+
+    public Integer getIntValue() {
+      return intValue;
+    }
+
+    public void setIntValue(Integer intValue) {
+      this.intValue = intValue;
+    }
+
+    public Boolean getBoolValue() {
+      return boolValue;
+    }
+
+    public void setBoolValue(Boolean boolValue) {
+      this.boolValue = boolValue;
+    }
+
+    public Map<String, String> getMapValue() {
+      return mapValue;
+    }
+
+    public void setMapValue(Map<String, String> mapValue) {
+      this.mapValue = mapValue;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      if (obj == this) {
+        return true;
+      }
+
+      if (!(obj instanceof JsonTestBean)) {
+        return false;
+      }
+
+      JsonTestBean that = (JsonTestBean) obj;
+      return Objects.equals(intValue, that.intValue)
+          && Objects.equals(boolValue, that.boolValue)
+          && Objects.equals(stringValue, that.stringValue)
+          && Objects.equals(mapValue, that.mapValue);
+    }
+  }
+}