diff --git a/submarine-server/server-submitter/submarine-k8s-agent/src/main/java/org/apache/submarine/server/k8s/agent/SubmarineAgentListener.java b/submarine-server/server-submitter/submarine-k8s-agent/src/main/java/org/apache/submarine/server/k8s/agent/SubmarineAgentListener.java
index 1ae7ec9c4..dd0a5992d 100644
--- a/submarine-server/server-submitter/submarine-k8s-agent/src/main/java/org/apache/submarine/server/k8s/agent/SubmarineAgentListener.java
+++ b/submarine-server/server-submitter/submarine-k8s-agent/src/main/java/org/apache/submarine/server/k8s/agent/SubmarineAgentListener.java
@@ -19,8 +19,6 @@
 
 package org.apache.submarine.server.k8s.agent;
 
-import io.fabric8.kubernetes.client.KubernetesClient;
-import io.fabric8.kubernetes.client.KubernetesClientBuilder;
 import io.javaoperatorsdk.operator.Operator;
 import io.javaoperatorsdk.operator.api.config.ControllerConfigurationOverrider;
 import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
@@ -49,10 +47,8 @@ public class SubmarineAgentListener {
   public static final DateTimeFormatter DTF = DateTimeFormatter.ISO_DATE_TIME;
 
   public static void main(String[] args) throws IOException {
-    // create kubernetes client
-    KubernetesClient client = new KubernetesClientBuilder().build();
     // create operator
-    Operator operator = new Operator(client);
+    Operator operator = new Operator();
     // scan all Reconciler implemented subclasses
     Reflections reflections = new Reflections("org.apache.submarine.server.k8s.agent");
     Set<Class<? extends Reconciler>> reconcilers = reflections.getSubTypesOf(Reconciler.class);
@@ -69,8 +65,6 @@ public static void main(String[] args) throws IOException {
         }
     );
     LOGGER.info("Starting agent with SUBMARINE_UID={}", OwnerReferenceConfig.getSubmarineUid());
-    // Adds a shutdown hook that automatically calls stop() when the app shuts down.
-    operator.installShutdownHook();
     // start operator
     operator.start();
     // Provide a lightweight service to handle health checks
diff --git a/submarine-server/server-submitter/submarine-k8s-agent/src/test/java/org/apache/submarine/server/k8s/agent/SubmitSubmarineAgentTest.java b/submarine-server/server-submitter/submarine-k8s-agent/src/test/java/org/apache/submarine/server/k8s/agent/SubmitSubmarineAgentTest.java
index fe20aed13..779bddfa0 100644
--- a/submarine-server/server-submitter/submarine-k8s-agent/src/test/java/org/apache/submarine/server/k8s/agent/SubmitSubmarineAgentTest.java
+++ b/submarine-server/server-submitter/submarine-k8s-agent/src/test/java/org/apache/submarine/server/k8s/agent/SubmitSubmarineAgentTest.java
@@ -24,9 +24,7 @@
 import io.fabric8.kubernetes.api.model.OwnerReferenceBuilder;
 import io.fabric8.kubernetes.api.model.apiextensions.v1.CustomResourceDefinition;
 import io.fabric8.kubernetes.client.KubernetesClient;
-import io.fabric8.kubernetes.client.dsl.base.CustomResourceDefinitionContext;
 import io.fabric8.kubernetes.client.server.mock.KubernetesServer;
-import io.fabric8.kubernetes.internal.KubernetesDeserializer;
 import io.javaoperatorsdk.operator.Operator;
 import org.apache.ibatis.session.SqlSession;
 import org.apache.submarine.commons.utils.SubmarineConfiguration;
@@ -103,10 +101,10 @@ public static void beforeInit() {
 
     // set client and operator
     client = server.getClient();
-    operator = new Operator(client);
+    operator = new Operator(client, null);
 
     // create notbook resource
-    KubernetesDeserializer.registerCustomKind("apiextensions.k8s.io/v1","Notebook", NotebookResource.class);
+    client.getKubernetesSerialization().registerKubernetesResource("apiextensions.k8s.io/v1","Notebook", NotebookResource.class);
     CustomResourceDefinition notebookCrd = client
             .apiextensions().v1()
             .customResourceDefinitions()
@@ -116,7 +114,7 @@ public static void beforeInit() {
     client.apiextensions().v1().customResourceDefinitions().createOrReplace(notebookCrd);
 
     // create tf resource
-    KubernetesDeserializer.registerCustomKind("apiextensions.k8s.io/v1", "TFJob", TFJob.class);
+    client.getKubernetesSerialization().registerKubernetesResource("apiextensions.k8s.io/v1", "TFJob", TFJob.class);
     CustomResourceDefinition tfCrd = client
             .apiextensions().v1()
             .customResourceDefinitions()
@@ -126,7 +124,7 @@ public static void beforeInit() {
     client.apiextensions().v1().customResourceDefinitions().create(tfCrd);
 
     // create pytorch resource
-    KubernetesDeserializer.registerCustomKind("apiextensions.k8s.io/v1", "PyTorchJob", PyTorchJob.class);
+    client.getKubernetesSerialization().registerKubernetesResource("apiextensions.k8s.io/v1", "PyTorchJob", PyTorchJob.class);
     CustomResourceDefinition ptCrd = client
             .apiextensions().v1()
             .customResourceDefinitions()
@@ -136,7 +134,7 @@ public static void beforeInit() {
     client.apiextensions().v1().customResourceDefinitions().create(ptCrd);
 
     // create xgboost resource
-    KubernetesDeserializer.registerCustomKind("apiextensions.k8s.io/v1", "XGBoostJob", XGBoostJob.class);
+    client.getKubernetesSerialization().registerKubernetesResource("apiextensions.k8s.io/v1", "XGBoostJob", XGBoostJob.class);
     CustomResourceDefinition xgbCrd = client
             .apiextensions().v1()
             .customResourceDefinitions()