diff --git a/sdk/python/kfp/dsl/_resource_op.py b/sdk/python/kfp/dsl/_resource_op.py
index b91878aae6c2..7ef76831ebb9 100644
--- a/sdk/python/kfp/dsl/_resource_op.py
+++ b/sdk/python/kfp/dsl/_resource_op.py
@@ -32,6 +32,7 @@ class Resource(object):
         "merge_strategy": "str",
         "success_condition": "str",
         "failure_condition": "str",
+        "set_owner_reference": "bool",
         "manifest": "str"
     }
     openapi_types = {
@@ -39,6 +40,7 @@ class Resource(object):
         "merge_strategy": "str",
         "success_condition": "str",
         "failure_condition": "str",
+        "set_owner_reference": "bool",
         "manifest": "str"
     }
     attribute_map = {
@@ -46,6 +48,7 @@ class Resource(object):
         "merge_strategy": "mergeStrategy",
         "success_condition": "successCondition",
         "failure_condition": "failureCondition",
+        "set_owner_reference": "setOwnerReference",
         "manifest": "manifest"
     }
 
@@ -54,12 +57,14 @@ def __init__(self,
                  merge_strategy: str = None,
                  success_condition: str = None,
                  failure_condition: str = None,
+                 set_owner_reference: bool = None,
                  manifest: str = None):
         """Create a new instance of Resource"""
         self.action = action
         self.merge_strategy = merge_strategy
         self.success_condition = success_condition
         self.failure_condition = failure_condition
+        self.set_owner_reference = set_owner_reference
         self.manifest = manifest
 
 
@@ -95,6 +100,7 @@ def __init__(self,
                  merge_strategy: str = None,
                  success_condition: str = None,
                  failure_condition: str = None,
+                 set_owner_reference: bool = None,
                  attribute_outputs: Dict[str, str] = None,
                  **kwargs):
 
@@ -121,7 +127,8 @@ def __init__(self,
             "action": action,
             "merge_strategy": merge_strategy,
             "success_condition": success_condition,
-            "failure_condition": failure_condition
+            "failure_condition": failure_condition,
+            "set_owner_reference": set_owner_reference,
         }
         # `resource` prop in `io.argoproj.workflow.v1alpha1.Template`
         self._resource = Resource(**init_resource)
diff --git a/sdk/python/kfp/dsl/_volume_op.py b/sdk/python/kfp/dsl/_volume_op.py
index d90351e3cb73..6a412da98fde 100644
--- a/sdk/python/kfp/dsl/_volume_op.py
+++ b/sdk/python/kfp/dsl/_volume_op.py
@@ -75,6 +75,9 @@ def __init__(self,
         # Add size to attribute outputs
         self.attribute_outputs = {"size": "{.status.capacity.storage}"}
 
+        if "set_owner_reference" not in kwargs:
+            kwargs["set_owner_reference"] = True
+
         if "k8s_resource" in kwargs:
             if resource_name or size or storage_class or modes or annotations:
                 raise ValueError("You cannot provide k8s_resource along with "
diff --git a/sdk/python/tests/compiler/testdata/param_substitutions.yaml b/sdk/python/tests/compiler/testdata/param_substitutions.yaml
index 76f1f1a9f8b7..e401302ee81f 100644
--- a/sdk/python/tests/compiler/testdata/param_substitutions.yaml
+++ b/sdk/python/tests/compiler/testdata/param_substitutions.yaml
@@ -44,6 +44,7 @@ spec:
       manifest: "apiVersion: v1\nkind: PersistentVolumeClaim\nmetadata:\n  name: '{{workflow.name}}-data'\n\
         spec:\n  accessModes:\n  - ReadWriteMany\n  resources:\n    requests:\n  \
         \    storage: 1Gi\n"
+      setOwnerReference: true
   - dag:
       tasks:
       - arguments:
diff --git a/sdk/python/tests/compiler/testdata/volume_snapshotop_rokurl.yaml b/sdk/python/tests/compiler/testdata/volume_snapshotop_rokurl.yaml
index 2d9e6570671b..c4212372780c 100644
--- a/sdk/python/tests/compiler/testdata/volume_snapshotop_rokurl.yaml
+++ b/sdk/python/tests/compiler/testdata/volume_snapshotop_rokurl.yaml
@@ -75,6 +75,7 @@ spec:
         \    rok/origin: '{{inputs.parameters.rok_url}}'\n  name: '{{workflow.name}}-vol1'\n\
         spec:\n  accessModes:\n  - ReadWriteMany\n  resources:\n    requests:\n  \
         \    storage: 1Gi\n"
+      setOwnerReference: true
   - inputs:
       parameters:
       - name: create-snapshot-1-name
@@ -97,6 +98,7 @@ spec:
         spec:\n  accessModes:\n  - ReadWriteMany\n  dataSource:\n    apiGroup: snapshot.storage.k8s.io\n\
         \    kind: VolumeSnapshot\n    name: '{{inputs.parameters.create-snapshot-1-name}}'\n\
         \  resources:\n    requests:\n      storage: '{{inputs.parameters.create-snapshot-1-size}}'\n"
+      setOwnerReference: true
   - inputs:
       parameters:
       - name: create-snapshot-2-name
@@ -119,6 +121,7 @@ spec:
         spec:\n  accessModes:\n  - ReadWriteMany\n  dataSource:\n    apiGroup: snapshot.storage.k8s.io\n\
         \    kind: VolumeSnapshot\n    name: '{{inputs.parameters.create-snapshot-2-name}}'\n\
         \  resources:\n    requests:\n      storage: '{{inputs.parameters.create-snapshot-2-size}}'\n"
+      setOwnerReference: true
   - container:
       args:
       - cat /data/file*| gzip -c >/data/full.gz
diff --git a/sdk/python/tests/compiler/testdata/volume_snapshotop_sequential.yaml b/sdk/python/tests/compiler/testdata/volume_snapshotop_sequential.yaml
index a0d9ac8f2abd..8cc9ed3cc3d5 100644
--- a/sdk/python/tests/compiler/testdata/volume_snapshotop_sequential.yaml
+++ b/sdk/python/tests/compiler/testdata/volume_snapshotop_sequential.yaml
@@ -29,6 +29,7 @@ spec:
       manifest: "apiVersion: v1\nkind: PersistentVolumeClaim\nmetadata:\n  name: '{{workflow.name}}-vol1'\n\
         spec:\n  accessModes:\n  - ReadWriteMany\n  resources:\n    requests:\n  \
         \    storage: 1Gi\n"
+      setOwnerReference: true
   - container:
       args:
       - mkdir /data/step1 && gsutil cat {{inputs.parameters.url}} | gzip -c >/data/step1/file1.gz
diff --git a/sdk/python/tests/compiler/testdata/volumeop_basic.yaml b/sdk/python/tests/compiler/testdata/volumeop_basic.yaml
index e2ab81035729..a378813eeabf 100644
--- a/sdk/python/tests/compiler/testdata/volumeop_basic.yaml
+++ b/sdk/python/tests/compiler/testdata/volumeop_basic.yaml
@@ -50,6 +50,7 @@ spec:
       manifest: "apiVersion: v1\nkind: PersistentVolumeClaim\nmetadata:\n  name: '{{workflow.name}}-my-pvc'\n\
         spec:\n  accessModes:\n  - ReadWriteMany\n  resources:\n    requests:\n  \
         \    storage: '{{inputs.parameters.size}}'\n"
+      setOwnerReference: true
   - dag:
       tasks:
       - arguments:
diff --git a/sdk/python/tests/compiler/testdata/volumeop_dag.yaml b/sdk/python/tests/compiler/testdata/volumeop_dag.yaml
index 420deff0e560..ad02d13889ef 100644
--- a/sdk/python/tests/compiler/testdata/volumeop_dag.yaml
+++ b/sdk/python/tests/compiler/testdata/volumeop_dag.yaml
@@ -28,6 +28,7 @@ spec:
       manifest: "apiVersion: v1\nkind: PersistentVolumeClaim\nmetadata:\n  name: '{{workflow.name}}-my-pvc'\n\
         spec:\n  accessModes:\n  - ReadWriteMany\n  resources:\n    requests:\n  \
         \    storage: 10Gi\n"
+      setOwnerReference: true
   - container:
       args:
       - echo 1 | tee /mnt/file1
diff --git a/sdk/python/tests/compiler/testdata/volumeop_parallel.yaml b/sdk/python/tests/compiler/testdata/volumeop_parallel.yaml
index cf278e506922..29c11d0b5e88 100644
--- a/sdk/python/tests/compiler/testdata/volumeop_parallel.yaml
+++ b/sdk/python/tests/compiler/testdata/volumeop_parallel.yaml
@@ -28,6 +28,7 @@ spec:
       manifest: "apiVersion: v1\nkind: PersistentVolumeClaim\nmetadata:\n  name: '{{workflow.name}}-my-pvc'\n\
         spec:\n  accessModes:\n  - ReadWriteMany\n  resources:\n    requests:\n  \
         \    storage: 10Gi\n"
+      setOwnerReference: true
   - container:
       args:
       - echo 1 | tee /mnt/file1
diff --git a/sdk/python/tests/compiler/testdata/volumeop_sequential.yaml b/sdk/python/tests/compiler/testdata/volumeop_sequential.yaml
index 0d9fe3727994..685b9857e196 100644
--- a/sdk/python/tests/compiler/testdata/volumeop_sequential.yaml
+++ b/sdk/python/tests/compiler/testdata/volumeop_sequential.yaml
@@ -28,6 +28,7 @@ spec:
       manifest: "apiVersion: v1\nkind: PersistentVolumeClaim\nmetadata:\n  name: '{{workflow.name}}-newpvc'\n\
         spec:\n  accessModes:\n  - ReadWriteMany\n  resources:\n    requests:\n  \
         \    storage: 10Gi\n"
+      setOwnerReference: true
   - container:
       args:
       - echo 1|tee /data/file1