diff --git a/docs/sphinx/user-docs/ray-cluster-interaction.rst b/docs/sphinx/user-docs/ray-cluster-interaction.rst
index 8e7929b4d..717f80678 100644
--- a/docs/sphinx/user-docs/ray-cluster-interaction.rst
+++ b/docs/sphinx/user-docs/ray-cluster-interaction.rst
@@ -66,6 +66,12 @@ cluster.up()
 
 | The ``cluster.up()`` function creates a Ray Cluster in the given namespace.
 
+cluster.apply()
+------------
+
+| The ``cluster.apply()`` function applies a Ray Cluster in the given namespace. If the cluster already exists, it is updated.
+| If it does not exist it is created.
+
 cluster.down()
 --------------
 
diff --git a/src/codeflare_sdk/ray/cluster/cluster.py b/src/codeflare_sdk/ray/cluster/cluster.py
index c6074afda..5324217be 100644
--- a/src/codeflare_sdk/ray/cluster/cluster.py
+++ b/src/codeflare_sdk/ray/cluster/cluster.py
@@ -59,6 +59,8 @@
 from kubernetes.client.rest import ApiException
 import warnings
 
+CF_SDK_FIELD_MANAGER = "codeflare-sdk"
+
 
 class Cluster:
     """
@@ -89,11 +91,9 @@ def __init__(self, config: ClusterConfiguration):
             cluster_up_down_buttons(self)
 
     def get_dynamic_client(self):  # pragma: no cover
-        """Return a dynamic client, optionally mocked in tests."""
         return DynamicClient(get_api_client())
 
     def config_check(self):
-        """Return a dynamic client, optionally mocked in tests."""
         return config_check()
 
     @property
@@ -201,7 +201,7 @@ def apply(self, force=False):
             crds = self.get_dynamic_client().resources
             if self.config.appwrapper:
                 api_version = "workload.codeflare.dev/v1beta2"
-                api_instance = crds.get(api_version, kind="AppWrapper")
+                api_instance = crds.get(api_version=api_version, kind="AppWrapper")
                 # defaulting body to resource_yaml
                 body = self.resource_yaml
                 if self.config.write_to_file:
@@ -210,6 +210,7 @@ def apply(self, force=False):
                         aw = yaml.load(f, Loader=yaml.FullLoader)
                     body = aw
                 api_instance.server_side_apply(
+                    field_manager=CF_SDK_FIELD_MANAGER,
                     group="workload.codeflare.dev",
                     version="v1beta2",
                     namespace=namespace,
@@ -218,9 +219,14 @@ def apply(self, force=False):
                 )
                 print(f"AppWrapper: '{name}' has successfully been created")
             else:
-                api_instance = crds.get(api_version="ray.io/v1", kind="RayCluster")
-                self._component_resources_apply(namespace, api_instance)
+                api_version = "ray.io/v1"
+                api_instance = crds.get(api_version=api_version, kind="RayCluster")
+                self._component_resources_apply(
+                    namespace=namespace, api_instance=api_instance
+                )
                 print(f"Ray Cluster: '{name}' has successfully been applied")
+        except AttributeError as e:
+            raise RuntimeError(f"Failed to initialize DynamicClient: {e}")
         except Exception as e:  # pragma: no cover
             return _kube_api_error_handling(e)
 
@@ -806,7 +812,7 @@ def _apply_resources(
     yamls, namespace: str, api_instance: client.CustomObjectsApi, force=False
 ):
     api_instance.server_side_apply(
-        field_manager="cluster-manager",
+        field_manager=CF_SDK_FIELD_MANAGER,
         group="ray.io",
         version="v1",
         namespace=namespace,