SUBMARINE-1251. fix patch type and add error handling

### What is this PR for?
In the previous PR #915, the patch method is not fixed since the `InvalidSpecException` is sending status code 200.

### What type of PR is it?
[Bug Fix]

### Todos

### What is the Jira issue?
https://issues.apache.org/jira/browse/SUBMARINE-1251

### How should this be tested?

### Screenshots (if appropriate)

### Questions:
* Do the license files need updating? No
* Are there breaking changes for older versions? No
* Does this need new documentation? No

Author: KUAN-HSUN-LI <b06209027@ntu.edu.tw>

Signed-off-by: Kevin <pingsutw@apache.org>

Closes #919 from KUAN-HSUN-LI/SUBMARINE-1251 and squashes the following commits:

2ce66f4 [KUAN-HSUN-LI] SUBMARINE-1251. fix patch type and add error handling

submarine-server/server-submitter/submitter-k8s/src/main/java/org/apache/submarine/server/submitter/k8s/K8sSubmitter.java

@@ -32,6 +32,7 @@
 import com.google.gson.Gson;
 import com.google.gson.JsonSyntaxException;
 
+import io.kubernetes.client.util.generic.options.PatchOptions;
 import okhttp3.OkHttpClient;
 import io.kubernetes.client.openapi.ApiClient;
 import io.kubernetes.client.openapi.ApiException;
@@ -318,19 +319,28 @@ public Experiment patchExperiment(ExperimentSpec spec) throws SubmarineRuntimeEx
     try {
       MLJob mlJob = ExperimentSpecParser.parseJob(spec);
       mlJob.getMetadata().setNamespace(getServerNamespace());
+      // Using apply yaml patch, field manager must be set, and it must be forced.
+      // https://kubernetes.io/docs/reference/using-api/server-side-apply/#field-management
+      PatchOptions patchOptions = new PatchOptions();
+      patchOptions.setFieldManager(spec.getMeta().getExperimentId());
+      patchOptions.setForce(true);
       Object object = mlJob.getPlural().equals(TFJob.CRD_TF_PLURAL_V1)
               ? tfJobClient.patch(getServerNamespace(), mlJob.getMetadata().getName(),
               V1Patch.PATCH_FORMAT_APPLY_YAML,
-              new V1Patch(new Gson().toJson(((TFJob) mlJob).getSpec()))).throwsApiException().getObject()
+              new V1Patch(new Gson().toJson(mlJob)),
+              patchOptions).throwsApiException().getObject()
               : pyTorchJobClient.patch(getServerNamespace(), mlJob.getMetadata().getName(),
               V1Patch.PATCH_FORMAT_APPLY_YAML,
-              new V1Patch(new Gson().toJson(((PyTorchJob) mlJob).getSpec()))).throwsApiException().getObject()
+              new V1Patch(new Gson().toJson(mlJob)),
+              patchOptions).throwsApiException().getObject()
               ;
       experiment = parseExperimentResponseObject(object, ParseOp.PARSE_OP_RESULT);
     } catch (InvalidSpecException e) {
-      throw new SubmarineRuntimeException(200, e.getMessage());
+      throw new SubmarineRuntimeException(409, e.getMessage());
     } catch (ApiException e) {
       throw new SubmarineRuntimeException(e.getCode(), e.getMessage());
+    } catch (Error e) {
+      throw new SubmarineRuntimeException(500, String.format("Unhandled error: %s", e.getMessage()));
     }
     return experiment;
   }