Fix #3906 - mount_pvc transform should ignore non-ContainerOps (#3912)

* Fix #3906 - check that ops to be transformed is a containerOp

* Update docstring for add_op_transformer to clarify that not only containerOp will be transformed.

sdk/python/kfp/dsl/_pipeline.py

@@ -102,9 +102,9 @@ def set_ttl_seconds_after_finished(self, seconds: int):
     """
     self.ttl_seconds_after_finished = seconds
     return self
-  
-  def set_default_pod_node_selector(self, label_name: str, value: str): 
-    """Add a constraint for nodeSelector for a pipeline. Each constraint is a key-value pair label. For the 
+
+  def set_default_pod_node_selector(self, label_name: str, value: str):
+    """Add a constraint for nodeSelector for a pipeline. Each constraint is a key-value pair label. For the
       container to be eligible to run on a node, the node must have each of the constraints appeared
       as labels.
 
@@ -114,7 +114,7 @@ def set_default_pod_node_selector(self, label_name: str, value: str):
     """
     self.default_pod_node_selector[label_name] = value
     return self
-  
+
 
   def set_image_pull_policy(self, policy: str):
     """Configures the default image pull policy
@@ -128,9 +128,10 @@ def set_image_pull_policy(self, policy: str):
 
   def add_op_transformer(self, transformer):
     """Configures the op_transformers which will be applied to all ops in the pipeline.
+    The ops can be ResourceOp, VolumenOp, or ContainerOp.
 
     Args:
-      transformer: a function that takes a ContainOp as input and returns a ContainerOp
+      transformer: a function that takes a kfp Op as input and returns a kfp Op
     """
     self.op_transformers.append(transformer)
 

sdk/python/kfp/onprem.py

@@ -1,14 +1,18 @@
 
 def mount_pvc(pvc_name='pipeline-claim', volume_name='pipeline', volume_mount_path='/mnt/pipeline'):
     """
-        Modifier function to apply to a Container Op to simplify volume, volume mount addition and 
+        Modifier function to apply to a Container Op to simplify volume, volume mount addition and
         enable better reuse of volumes, volume claims across container ops.
         Usage:
             train = train_op(...)
             train.apply(mount_pvc('claim-name', 'pipeline', '/mnt/pipeline'))
     """
     def _mount_pvc(task):
         from kubernetes import client as k8s_client
+        # there can be other ops in a pipeline (e.g. ResourceOp, VolumeOp)
+        # refer to #3906
+        if not hasattr(task, "add_volume") or not hasattr(task, "add_volume_mount"):
+            return task
         local_pvc = k8s_client.V1PersistentVolumeClaimVolumeSource(claim_name=pvc_name)
         return (
             task