We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Crate VolumeOP from  k8s_resource error
**What happened:**
when I run
```
 dsl.VolumeOP(k8s_resource=my_vpc)
```

```
 Raises:
            ValueError: if k8s_resource is provided along with other arguments
```


I think the reason is :
``` 
    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 "
                                 "other arguments.")
```
```
    def __init__(self,
                 resource_name: str = None,
                 size: str = None,
                 storage_class: str = None,
                 modes: List[str] = VOLUME_MODE_RWM,
                 annotations: Dict[str, str] = None,
                 data_source=None,
                 **kwargs):
```

but the mode has a default value
sdk/python/kfp/dsl/_volume_op.py
**What did you expect to happen:**
I think, I should only put k8s_resource and name in it.
**What steps did you take:**
[A clear and concise description of what the bug is.]

**Anything else you would like to add:**
[Miscellaneous information that will assist in solving the issue.]



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sdk/python/kfp/dsl/_volume_op.py`
Content:
```
1 # Copyright 2019 Google LLC
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 
16 import re
17 from typing import List, Dict
18 from kubernetes.client.models import (
19     V1ObjectMeta, V1ResourceRequirements, V1PersistentVolumeClaimSpec,
20     V1PersistentVolumeClaim, V1TypedLocalObjectReference
21 )
22 
23 from ._resource_op import ResourceOp
24 from ._pipeline_param import (
25     PipelineParam, match_serialized_pipelineparam, sanitize_k8s_name
26 )
27 from ._pipeline_volume import PipelineVolume
28 
29 
30 VOLUME_MODE_RWO = ["ReadWriteOnce"]
31 VOLUME_MODE_RWM = ["ReadWriteMany"]
32 VOLUME_MODE_ROM = ["ReadOnlyMany"]
33 
34 
35 class VolumeOp(ResourceOp):
36     """Represents an op which will be translated into a resource template
37     which will be creating a PVC.
38     """
39 
40     def __init__(self,
41                  resource_name: str = None,
42                  size: str = None,
43                  storage_class: str = None,
44                  modes: List[str] = VOLUME_MODE_RWM,
45                  annotations: Dict[str, str] = None,
46                  data_source=None,
47                  **kwargs):
48         """Create a new instance of VolumeOp.
49 
50         Args:
51             resource_name: A desired name for the PVC which will be created
52             size: The size of the PVC which will be created
53             storage_class: The storage class to use for the dynamically created
54                 PVC
55             modes: The access modes for the PVC
56             annotations: Annotations to be patched in the PVC
57             data_source: May be a V1TypedLocalObjectReference, and then it is
58                 used in the data_source field of the PVC as is. Can also be a
59                 string/PipelineParam, and in that case it will be used as a
60                 VolumeSnapshot name (Alpha feature)
61             kwargs: See ResourceOp definition
62         Raises:
63             ValueError: if k8s_resource is provided along with other arguments
64                         if k8s_resource is not a V1PersistentVolumeClaim
65                         if size is None
66                         if size is an invalid memory string (when not a
67                             PipelineParam)
68                         if data_source is not one of (str, PipelineParam,
69                             V1TypedLocalObjectReference)
70         """
71         # Add size to attribute outputs
72         self.attribute_outputs = {"size": "{.status.capacity.storage}"}
73 
74         if "k8s_resource" in kwargs:
75             if resource_name or size or storage_class or modes or annotations:
76                 raise ValueError("You cannot provide k8s_resource along with "
77                                  "other arguments.")
78             if not isinstance(kwargs["k8s_resource"], V1PersistentVolumeClaim):
79                 raise ValueError("k8s_resource in VolumeOp must be an instance"
80                                  " of V1PersistentVolumeClaim")
81             super().__init__(**kwargs)
82             self.volume = PipelineVolume(
83                 name=sanitize_k8s_name(self.name),
84                 pvc=self.outputs["name"]
85             )
86             return
87 
88         if not size:
89             raise ValueError("Please provide size")
90         elif not match_serialized_pipelineparam(str(size)):
91             self._validate_memory_string(size)
92 
93         if data_source and not isinstance(
94            data_source, (str, PipelineParam, V1TypedLocalObjectReference)):
95             raise ValueError("data_source can be one of (str, PipelineParam, "
96                              "V1TypedLocalObjectReference).")
97         if data_source and isinstance(data_source, (str, PipelineParam)):
98             data_source = V1TypedLocalObjectReference(
99                 api_group="snapshot.storage.k8s.io",
100                 kind="VolumeSnapshot",
101                 name=data_source
102             )
103 
104         # Set the k8s_resource
105         if not match_serialized_pipelineparam(str(resource_name)):
106             resource_name = sanitize_k8s_name(resource_name)
107         pvc_metadata = V1ObjectMeta(
108             name="{{workflow.name}}-%s" % resource_name,
109             annotations=annotations
110         )
111         requested_resources = V1ResourceRequirements(
112             requests={"storage": size}
113         )
114         pvc_spec = V1PersistentVolumeClaimSpec(
115             access_modes=modes,
116             resources=requested_resources,
117             storage_class_name=storage_class,
118             data_source=data_source
119         )
120         k8s_resource = V1PersistentVolumeClaim(
121             api_version="v1",
122             kind="PersistentVolumeClaim",
123             metadata=pvc_metadata,
124             spec=pvc_spec
125         )
126 
127         super().__init__(
128             k8s_resource=k8s_resource,
129             **kwargs,
130         )
131         self.volume = PipelineVolume(
132             name=sanitize_k8s_name(self.name),
133             pvc=self.outputs["name"]
134         )
135 
136     def _validate_memory_string(self, memory_string):
137         """Validate a given string is valid for memory request or limit."""
138         if re.match(r'^[0-9]+(E|Ei|P|Pi|T|Ti|G|Gi|M|Mi|K|Ki){0,1}$',
139                     memory_string) is None:
140             raise ValueError('Invalid memory string. Should be an integer, ' +
141                              'or integer followed by one of ' +
142                              '"E|Ei|P|Pi|T|Ti|G|Gi|M|Mi|K|Ki"')
143
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/sdk/python/kfp/dsl/_volume_op.py b/sdk/python/kfp/dsl/_volume_op.py
--- a/sdk/python/kfp/dsl/_volume_op.py
+++ b/sdk/python/kfp/dsl/_volume_op.py
@@ -41,7 +41,7 @@
                  resource_name: str = None,
                  size: str = None,
                  storage_class: str = None,
-                 modes: List[str] = VOLUME_MODE_RWM,
+                 modes: List[str] = None,
                  annotations: Dict[str, str] = None,
                  data_source=None,
                  **kwargs):
@@ -112,7 +112,7 @@
             requests={"storage": size}
         )
         pvc_spec = V1PersistentVolumeClaimSpec(
-            access_modes=modes,
+            access_modes=modes or VOLUME_MODE_RWM,
             resources=requested_resources,
             storage_class_name=storage_class,
             data_source=data_source

{"golden_diff": "diff --git a/sdk/python/kfp/dsl/_volume_op.py b/sdk/python/kfp/dsl/_volume_op.py\n--- a/sdk/python/kfp/dsl/_volume_op.py\n+++ b/sdk/python/kfp/dsl/_volume_op.py\n@@ -41,7 +41,7 @@\n                  resource_name: str = None,\n                  size: str = None,\n                  storage_class: str = None,\n-                 modes: List[str] = VOLUME_MODE_RWM,\n+                 modes: List[str] = None,\n                  annotations: Dict[str, str] = None,\n                  data_source=None,\n                  **kwargs):\n@@ -112,7 +112,7 @@\n             requests={\"storage\": size}\n         )\n         pvc_spec = V1PersistentVolumeClaimSpec(\n-            access_modes=modes,\n+            access_modes=modes or VOLUME_MODE_RWM,\n             resources=requested_resources,\n             storage_class_name=storage_class,\n             data_source=data_source\n", "issue": "Crate VolumeOP from  k8s_resource error\n**What happened:**\r\nwhen I run\r\n```\r\n dsl.VolumeOP(k8s_resource=my_vpc)\r\n```\r\n\r\n```\r\n Raises:\r\n            ValueError: if k8s_resource is provided along with other arguments\r\n```\r\n\r\n\r\nI think the reason is :\r\n``` \r\n    if \"k8s_resource\" in kwargs:\r\n            if resource_name or size or storage_class or modes or annotations:\r\n                raise ValueError(\"You cannot provide k8s_resource along with \"\r\n                                 \"other arguments.\")\r\n```\r\n```\r\n    def __init__(self,\r\n                 resource_name: str = None,\r\n                 size: str = None,\r\n                 storage_class: str = None,\r\n                 modes: List[str] = VOLUME_MODE_RWM,\r\n                 annotations: Dict[str, str] = None,\r\n                 data_source=None,\r\n                 **kwargs):\r\n```\r\n\r\nbut the mode has a default value\r\nsdk/python/kfp/dsl/_volume_op.py\r\n**What did you expect to happen:**\r\nI think, I should only put k8s_resource and name in it.\r\n**What steps did you take:**\r\n[A clear and concise description of what the bug is.]\r\n\r\n**Anything else you would like to add:**\r\n[Miscellaneous information that will assist in solving the issue.]\r\n\r\n\n", "before_files": [{"content": "# Copyright 2019 Google LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\nimport re\nfrom typing import List, Dict\nfrom kubernetes.client.models import (\n    V1ObjectMeta, V1ResourceRequirements, V1PersistentVolumeClaimSpec,\n    V1PersistentVolumeClaim, V1TypedLocalObjectReference\n)\n\nfrom ._resource_op import ResourceOp\nfrom ._pipeline_param import (\n    PipelineParam, match_serialized_pipelineparam, sanitize_k8s_name\n)\nfrom ._pipeline_volume import PipelineVolume\n\n\nVOLUME_MODE_RWO = [\"ReadWriteOnce\"]\nVOLUME_MODE_RWM = [\"ReadWriteMany\"]\nVOLUME_MODE_ROM = [\"ReadOnlyMany\"]\n\n\nclass VolumeOp(ResourceOp):\n    \"\"\"Represents an op which will be translated into a resource template\n    which will be creating a PVC.\n    \"\"\"\n\n    def __init__(self,\n                 resource_name: str = None,\n                 size: str = None,\n                 storage_class: str = None,\n                 modes: List[str] = VOLUME_MODE_RWM,\n                 annotations: Dict[str, str] = None,\n                 data_source=None,\n                 **kwargs):\n        \"\"\"Create a new instance of VolumeOp.\n\n        Args:\n            resource_name: A desired name for the PVC which will be created\n            size: The size of the PVC which will be created\n            storage_class: The storage class to use for the dynamically created\n                PVC\n            modes: The access modes for the PVC\n            annotations: Annotations to be patched in the PVC\n            data_source: May be a V1TypedLocalObjectReference, and then it is\n                used in the data_source field of the PVC as is. Can also be a\n                string/PipelineParam, and in that case it will be used as a\n                VolumeSnapshot name (Alpha feature)\n            kwargs: See ResourceOp definition\n        Raises:\n            ValueError: if k8s_resource is provided along with other arguments\n                        if k8s_resource is not a V1PersistentVolumeClaim\n                        if size is None\n                        if size is an invalid memory string (when not a\n                            PipelineParam)\n                        if data_source is not one of (str, PipelineParam,\n                            V1TypedLocalObjectReference)\n        \"\"\"\n        # Add size to attribute outputs\n        self.attribute_outputs = {\"size\": \"{.status.capacity.storage}\"}\n\n        if \"k8s_resource\" in kwargs:\n            if resource_name or size or storage_class or modes or annotations:\n                raise ValueError(\"You cannot provide k8s_resource along with \"\n                                 \"other arguments.\")\n            if not isinstance(kwargs[\"k8s_resource\"], V1PersistentVolumeClaim):\n                raise ValueError(\"k8s_resource in VolumeOp must be an instance\"\n                                 \" of V1PersistentVolumeClaim\")\n            super().__init__(**kwargs)\n            self.volume = PipelineVolume(\n                name=sanitize_k8s_name(self.name),\n                pvc=self.outputs[\"name\"]\n            )\n            return\n\n        if not size:\n            raise ValueError(\"Please provide size\")\n        elif not match_serialized_pipelineparam(str(size)):\n            self._validate_memory_string(size)\n\n        if data_source and not isinstance(\n           data_source, (str, PipelineParam, V1TypedLocalObjectReference)):\n            raise ValueError(\"data_source can be one of (str, PipelineParam, \"\n                             \"V1TypedLocalObjectReference).\")\n        if data_source and isinstance(data_source, (str, PipelineParam)):\n            data_source = V1TypedLocalObjectReference(\n                api_group=\"snapshot.storage.k8s.io\",\n                kind=\"VolumeSnapshot\",\n                name=data_source\n            )\n\n        # Set the k8s_resource\n        if not match_serialized_pipelineparam(str(resource_name)):\n            resource_name = sanitize_k8s_name(resource_name)\n        pvc_metadata = V1ObjectMeta(\n            name=\"{{workflow.name}}-%s\" % resource_name,\n            annotations=annotations\n        )\n        requested_resources = V1ResourceRequirements(\n            requests={\"storage\": size}\n        )\n        pvc_spec = V1PersistentVolumeClaimSpec(\n            access_modes=modes,\n            resources=requested_resources,\n            storage_class_name=storage_class,\n            data_source=data_source\n        )\n        k8s_resource = V1PersistentVolumeClaim(\n            api_version=\"v1\",\n            kind=\"PersistentVolumeClaim\",\n            metadata=pvc_metadata,\n            spec=pvc_spec\n        )\n\n        super().__init__(\n            k8s_resource=k8s_resource,\n            **kwargs,\n        )\n        self.volume = PipelineVolume(\n            name=sanitize_k8s_name(self.name),\n            pvc=self.outputs[\"name\"]\n        )\n\n    def _validate_memory_string(self, memory_string):\n        \"\"\"Validate a given string is valid for memory request or limit.\"\"\"\n        if re.match(r'^[0-9]+(E|Ei|P|Pi|T|Ti|G|Gi|M|Mi|K|Ki){0,1}$',\n                    memory_string) is None:\n            raise ValueError('Invalid memory string. Should be an integer, ' +\n                             'or integer followed by one of ' +\n                             '\"E|Ei|P|Pi|T|Ti|G|Gi|M|Mi|K|Ki\"')\n", "path": "sdk/python/kfp/dsl/_volume_op.py"}], "after_files": [{"content": "# Copyright 2019 Google LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\nimport re\nfrom typing import List, Dict\nfrom kubernetes.client.models import (\n    V1ObjectMeta, V1ResourceRequirements, V1PersistentVolumeClaimSpec,\n    V1PersistentVolumeClaim, V1TypedLocalObjectReference\n)\n\nfrom ._resource_op import ResourceOp\nfrom ._pipeline_param import (\n    PipelineParam, match_serialized_pipelineparam, sanitize_k8s_name\n)\nfrom ._pipeline_volume import PipelineVolume\n\n\nVOLUME_MODE_RWO = [\"ReadWriteOnce\"]\nVOLUME_MODE_RWM = [\"ReadWriteMany\"]\nVOLUME_MODE_ROM = [\"ReadOnlyMany\"]\n\n\nclass VolumeOp(ResourceOp):\n    \"\"\"Represents an op which will be translated into a resource template\n    which will be creating a PVC.\n    \"\"\"\n\n    def __init__(self,\n                 resource_name: str = None,\n                 size: str = None,\n                 storage_class: str = None,\n                 modes: List[str] = None,\n                 annotations: Dict[str, str] = None,\n                 data_source=None,\n                 **kwargs):\n        \"\"\"Create a new instance of VolumeOp.\n\n        Args:\n            resource_name: A desired name for the PVC which will be created\n            size: The size of the PVC which will be created\n            storage_class: The storage class to use for the dynamically created\n                PVC\n            modes: The access modes for the PVC\n            annotations: Annotations to be patched in the PVC\n            data_source: May be a V1TypedLocalObjectReference, and then it is\n                used in the data_source field of the PVC as is. Can also be a\n                string/PipelineParam, and in that case it will be used as a\n                VolumeSnapshot name (Alpha feature)\n            kwargs: See ResourceOp definition\n        Raises:\n            ValueError: if k8s_resource is provided along with other arguments\n                        if k8s_resource is not a V1PersistentVolumeClaim\n                        if size is None\n                        if size is an invalid memory string (when not a\n                            PipelineParam)\n                        if data_source is not one of (str, PipelineParam,\n                            V1TypedLocalObjectReference)\n        \"\"\"\n        # Add size to attribute outputs\n        self.attribute_outputs = {\"size\": \"{.status.capacity.storage}\"}\n\n        if \"k8s_resource\" in kwargs:\n            if resource_name or size or storage_class or modes or annotations:\n                raise ValueError(\"You cannot provide k8s_resource along with \"\n                                 \"other arguments.\")\n            if not isinstance(kwargs[\"k8s_resource\"], V1PersistentVolumeClaim):\n                raise ValueError(\"k8s_resource in VolumeOp must be an instance\"\n                                 \" of V1PersistentVolumeClaim\")\n            super().__init__(**kwargs)\n            self.volume = PipelineVolume(\n                name=sanitize_k8s_name(self.name),\n                pvc=self.outputs[\"name\"]\n            )\n            return\n\n        if not size:\n            raise ValueError(\"Please provide size\")\n        elif not match_serialized_pipelineparam(str(size)):\n            self._validate_memory_string(size)\n\n        if data_source and not isinstance(\n           data_source, (str, PipelineParam, V1TypedLocalObjectReference)):\n            raise ValueError(\"data_source can be one of (str, PipelineParam, \"\n                             \"V1TypedLocalObjectReference).\")\n        if data_source and isinstance(data_source, (str, PipelineParam)):\n            data_source = V1TypedLocalObjectReference(\n                api_group=\"snapshot.storage.k8s.io\",\n                kind=\"VolumeSnapshot\",\n                name=data_source\n            )\n\n        # Set the k8s_resource\n        if not match_serialized_pipelineparam(str(resource_name)):\n            resource_name = sanitize_k8s_name(resource_name)\n        pvc_metadata = V1ObjectMeta(\n            name=\"{{workflow.name}}-%s\" % resource_name,\n            annotations=annotations\n        )\n        requested_resources = V1ResourceRequirements(\n            requests={\"storage\": size}\n        )\n        pvc_spec = V1PersistentVolumeClaimSpec(\n            access_modes=modes or VOLUME_MODE_RWM,\n            resources=requested_resources,\n            storage_class_name=storage_class,\n            data_source=data_source\n        )\n        k8s_resource = V1PersistentVolumeClaim(\n            api_version=\"v1\",\n            kind=\"PersistentVolumeClaim\",\n            metadata=pvc_metadata,\n            spec=pvc_spec\n        )\n\n        super().__init__(\n            k8s_resource=k8s_resource,\n            **kwargs,\n        )\n        self.volume = PipelineVolume(\n            name=sanitize_k8s_name(self.name),\n            pvc=self.outputs[\"name\"]\n        )\n\n    def _validate_memory_string(self, memory_string):\n        \"\"\"Validate a given string is valid for memory request or limit.\"\"\"\n        if re.match(r'^[0-9]+(E|Ei|P|Pi|T|Ti|G|Gi|M|Mi|K|Ki){0,1}$',\n                    memory_string) is None:\n            raise ValueError('Invalid memory string. Should be an integer, ' +\n                             'or integer followed by one of ' +\n                             '\"E|Ei|P|Pi|T|Ti|G|Gi|M|Mi|K|Ki\"')\n", "path": "sdk/python/kfp/dsl/_volume_op.py"}]}