🔧 use application/json-patch+json when patching k8s objects

src/dagster_ray/kuberay/client/base.py

@@ -161,6 +161,18 @@ def update(self, name: str, namespace: str, body: Any):
             namespace=namespace,
         )
 
+    def update_json_patch(self, name: str, namespace: str, body: Any) -> Any:
+        data, status_code, headers = self._api.api_client.call_api(  # pyright: ignore[reportGeneralTypeIssues]
+            f"/apis/{self.group}/{self.version}/namespaces/{namespace}/{self.plural}/{name}",
+            "PATCH",
+            header_params={"Content-Type": "application/json-patch+json"},
+            body=body,
+            response_type="object",
+            _preload_content=True,
+            async_req=False,
+        )
+        return data
+
     def wait_until_exists(self, name: str, namespace: str, timeout: float = 60.0, poll_interval: float = 1.0) -> None:
         from kubernetes.client import ApiException
 

src/dagster_ray/kuberay/resources/raycluster.py

@@ -129,11 +129,21 @@ def create(self, context: AnyDagsterContext):
                 self._name = cluster_name
 
                 # place a lock on the cluster
-
-                self.client.update(
+                # Create individual patch operations for each annotation to avoid replacing existing ones
+                lock_annotations = self.get_sharing_lock_annotations(context)
+                patch_operations = [
+                    {
+                        "op": "add",
+                        "path": f"/metadata/annotations/{key.replace('/', '~1')}",
+                        "value": value,
+                    }
+                    for key, value in lock_annotations.items()
+                ]
+
+                self.client.update_json_patch(
                     name=cluster_name,
                     namespace=self.namespace,
-                    body={"metadata": {"annotations": self.get_sharing_lock_annotations(context)}},
+                    body=patch_operations,
                 )
 
                 return

src/dagster_ray/kuberay/resources/rayjob.py

@@ -179,10 +179,10 @@ def connect(self, context: AnyDagsterContext) -> "RayBaseContext":
         ray_context = super().connect(context)
 
         # now point the RayJob at our ray job
-        self.client.update(
+        self.client.update_json_patch(
             name=self.name,
             namespace=self.namespace,
-            body={"spec": {"jobId": self.runtime_job_id}},
+            body=[{"op": "replace", "path": "/spec/jobId", "value": self.runtime_job_id}],
         )
 
         return ray_context