We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
allow output artifact store configuration (vs hard coded)
it seems like the output artifacts are always stored in a specific minio service, port, namespace, bucket, secrets, etc (`minio-service.kubeflow:9000`). 

see: https://github.com/kubeflow/pipelines/blob/f40a22a3f4a8e06d20cf3e3f425b5058d5c87e0b/sdk/python/kfp/compiler/_op_to_template.py#L148

it would be great to make it flexible, e.g. allow using S3, or change namespace or bucket names.
i suggest making it configurable, i can do such PR if we agree its needed. 
flexible pipeline service (host) path in client SDK 
when creating an SDK `Client()` the path to `ml-pipeline` API service is loaded from a hard coded value (`ml-pipeline.kubeflow.svc.cluster.local:8888`) which indicate a specific k8s namespace. it can be valuable to load that default value from an env variable, i.e. changing the line in `_client.py` from:

`config.host = host if host else Client.IN_CLUSTER_DNS_NAME`

to:

`config.host = host or os.environ.get('ML_PIPELINE_DNS_NAME',Client.IN_CLUSTER_DNS_NAME)`

also note that when a user provide the `host` parameter, the ipython output points to the API server and not to the UI service (see the logic in `_get_url_prefix()`), it seems like a potential bug

if its acceptable i can submit a PR for the line change above
 

--- 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/_component_bridge.py`
Content:
```
1 # Copyright 2018 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 import copy
16 from typing import Any, Mapping
17 from ..components.structures import ComponentSpec, ComponentReference
18 from ..components._components import _default_component_name, _resolve_command_line_and_paths
19 from ..components._naming import _sanitize_python_function_name, generate_unique_name_conversion_table
20 from .. import dsl
21 
22 
23 def _create_container_op_from_component_and_arguments(
24     component_spec: ComponentSpec,
25     arguments: Mapping[str, Any],
26     component_ref: ComponentReference = None,
27 ) -> 'dsl.ContainerOp':
28     # Check types of the reference arguments and serialize PipelineParams
29     arguments = arguments.copy()
30     for input_name, argument_value in arguments.items():
31         if isinstance(argument_value, dsl.PipelineParam):
32             input_type = component_spec._inputs_dict[input_name].type
33             reference_type = argument_value.param_type
34             dsl.types.verify_type_compatibility(reference_type, input_type, 'Incompatible argument passed to the input "{}" of component "{}": '.format(input_name, component_spec.name))
35 
36             arguments[input_name] = str(argument_value)
37 
38     resolved_cmd = _resolve_command_line_and_paths(
39         component_spec=component_spec,
40         arguments=arguments,
41     )
42 
43     container_spec = component_spec.implementation.container
44 
45     task = dsl.ContainerOp(
46         name=component_spec.name or _default_component_name,
47         image=container_spec.image,
48         command=resolved_cmd.command,
49         arguments=resolved_cmd.args,
50         file_outputs=resolved_cmd.output_paths,
51         artifact_argument_paths=[
52             dsl.InputArgumentPath(
53                 argument=arguments[input_name],
54                 input=input_name,
55                 path=path,
56             )
57             for input_name, path in resolved_cmd.input_paths.items()
58         ],
59     )
60 
61     component_meta = copy.copy(component_spec)
62     task._set_metadata(component_meta)
63     component_ref_without_spec = copy.copy(component_ref)
64     component_ref_without_spec.spec = None
65     task._component_ref = component_ref_without_spec
66 
67     # Previously, ContainerOp had strict requirements for the output names, so we had to
68     # convert all the names before passing them to the ContainerOp constructor.
69     # Outputs with non-pythonic names could not be accessed using their original names.
70     # Now ContainerOp supports any output names, so we're now using the original output names.
71     # However to support legacy pipelines, we're also adding output references with pythonic names.
72     # TODO: Add warning when people use the legacy output names.
73     output_names = [output_spec.name for output_spec in component_spec.outputs or []] # Stabilizing the ordering
74     output_name_to_python = generate_unique_name_conversion_table(output_names, _sanitize_python_function_name)
75     for output_name in output_names:
76         pythonic_output_name = output_name_to_python[output_name]
77         # Note: Some component outputs are currently missing from task.outputs (e.g. MLPipeline UI Metadata)
78         if pythonic_output_name not in task.outputs and output_name in task.outputs:
79             task.outputs[pythonic_output_name] = task.outputs[output_name]
80 
81     if container_spec.env:
82         from kubernetes import client as k8s_client
83         for name, value in container_spec.env.items():
84             task.container.add_env_variable(k8s_client.V1EnvVar(name=name, value=value))
85 
86     if component_spec.metadata:
87         for key, value in (component_spec.metadata.annotations or {}).items():
88             task.add_pod_annotation(key, value)
89         for key, value in (component_spec.metadata.labels or {}).items():
90             task.add_pod_label(key, value)
91 
92     return task
93
```
--- 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/_component_bridge.py b/sdk/python/kfp/dsl/_component_bridge.py
--- a/sdk/python/kfp/dsl/_component_bridge.py
+++ b/sdk/python/kfp/dsl/_component_bridge.py
@@ -84,9 +84,13 @@
             task.container.add_env_variable(k8s_client.V1EnvVar(name=name, value=value))
 
     if component_spec.metadata:
-        for key, value in (component_spec.metadata.annotations or {}).items():
+        annotations = component_spec.metadata.annotations or {}
+        for key, value in annotations.items():
             task.add_pod_annotation(key, value)
         for key, value in (component_spec.metadata.labels or {}).items():
             task.add_pod_label(key, value)
+        # Disabling the caching for the volatile components by default
+        if annotations.get('volatile_component', 'false') == 'true':
+            task.execution_options.caching_strategy.max_cache_staleness = 'P0D'
 
     return task

{"golden_diff": "diff --git a/sdk/python/kfp/dsl/_component_bridge.py b/sdk/python/kfp/dsl/_component_bridge.py\n--- a/sdk/python/kfp/dsl/_component_bridge.py\n+++ b/sdk/python/kfp/dsl/_component_bridge.py\n@@ -84,9 +84,13 @@\n             task.container.add_env_variable(k8s_client.V1EnvVar(name=name, value=value))\n \n     if component_spec.metadata:\n-        for key, value in (component_spec.metadata.annotations or {}).items():\n+        annotations = component_spec.metadata.annotations or {}\n+        for key, value in annotations.items():\n             task.add_pod_annotation(key, value)\n         for key, value in (component_spec.metadata.labels or {}).items():\n             task.add_pod_label(key, value)\n+        # Disabling the caching for the volatile components by default\n+        if annotations.get('volatile_component', 'false') == 'true':\n+            task.execution_options.caching_strategy.max_cache_staleness = 'P0D'\n \n     return task\n", "issue": "allow output artifact store configuration (vs hard coded)\nit seems like the output artifacts are always stored in a specific minio service, port, namespace, bucket, secrets, etc (`minio-service.kubeflow:9000`). \r\n\r\nsee: https://github.com/kubeflow/pipelines/blob/f40a22a3f4a8e06d20cf3e3f425b5058d5c87e0b/sdk/python/kfp/compiler/_op_to_template.py#L148\r\n\r\nit would be great to make it flexible, e.g. allow using S3, or change namespace or bucket names.\r\ni suggest making it configurable, i can do such PR if we agree its needed. \nflexible pipeline service (host) path in client SDK \nwhen creating an SDK `Client()` the path to `ml-pipeline` API service is loaded from a hard coded value (`ml-pipeline.kubeflow.svc.cluster.local:8888`) which indicate a specific k8s namespace. it can be valuable to load that default value from an env variable, i.e. changing the line in `_client.py` from:\r\n\r\n`config.host = host if host else Client.IN_CLUSTER_DNS_NAME`\r\n\r\nto:\r\n\r\n`config.host = host or os.environ.get('ML_PIPELINE_DNS_NAME',Client.IN_CLUSTER_DNS_NAME)`\r\n\r\nalso note that when a user provide the `host` parameter, the ipython output points to the API server and not to the UI service (see the logic in `_get_url_prefix()`), it seems like a potential bug\r\n\r\nif its acceptable i can submit a PR for the line change above\r\n \n", "before_files": [{"content": "# Copyright 2018 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\nimport copy\nfrom typing import Any, Mapping\nfrom ..components.structures import ComponentSpec, ComponentReference\nfrom ..components._components import _default_component_name, _resolve_command_line_and_paths\nfrom ..components._naming import _sanitize_python_function_name, generate_unique_name_conversion_table\nfrom .. import dsl\n\n\ndef _create_container_op_from_component_and_arguments(\n    component_spec: ComponentSpec,\n    arguments: Mapping[str, Any],\n    component_ref: ComponentReference = None,\n) -> 'dsl.ContainerOp':\n    # Check types of the reference arguments and serialize PipelineParams\n    arguments = arguments.copy()\n    for input_name, argument_value in arguments.items():\n        if isinstance(argument_value, dsl.PipelineParam):\n            input_type = component_spec._inputs_dict[input_name].type\n            reference_type = argument_value.param_type\n            dsl.types.verify_type_compatibility(reference_type, input_type, 'Incompatible argument passed to the input \"{}\" of component \"{}\": '.format(input_name, component_spec.name))\n\n            arguments[input_name] = str(argument_value)\n\n    resolved_cmd = _resolve_command_line_and_paths(\n        component_spec=component_spec,\n        arguments=arguments,\n    )\n\n    container_spec = component_spec.implementation.container\n\n    task = dsl.ContainerOp(\n        name=component_spec.name or _default_component_name,\n        image=container_spec.image,\n        command=resolved_cmd.command,\n        arguments=resolved_cmd.args,\n        file_outputs=resolved_cmd.output_paths,\n        artifact_argument_paths=[\n            dsl.InputArgumentPath(\n                argument=arguments[input_name],\n                input=input_name,\n                path=path,\n            )\n            for input_name, path in resolved_cmd.input_paths.items()\n        ],\n    )\n\n    component_meta = copy.copy(component_spec)\n    task._set_metadata(component_meta)\n    component_ref_without_spec = copy.copy(component_ref)\n    component_ref_without_spec.spec = None\n    task._component_ref = component_ref_without_spec\n\n    # Previously, ContainerOp had strict requirements for the output names, so we had to\n    # convert all the names before passing them to the ContainerOp constructor.\n    # Outputs with non-pythonic names could not be accessed using their original names.\n    # Now ContainerOp supports any output names, so we're now using the original output names.\n    # However to support legacy pipelines, we're also adding output references with pythonic names.\n    # TODO: Add warning when people use the legacy output names.\n    output_names = [output_spec.name for output_spec in component_spec.outputs or []] # Stabilizing the ordering\n    output_name_to_python = generate_unique_name_conversion_table(output_names, _sanitize_python_function_name)\n    for output_name in output_names:\n        pythonic_output_name = output_name_to_python[output_name]\n        # Note: Some component outputs are currently missing from task.outputs (e.g. MLPipeline UI Metadata)\n        if pythonic_output_name not in task.outputs and output_name in task.outputs:\n            task.outputs[pythonic_output_name] = task.outputs[output_name]\n\n    if container_spec.env:\n        from kubernetes import client as k8s_client\n        for name, value in container_spec.env.items():\n            task.container.add_env_variable(k8s_client.V1EnvVar(name=name, value=value))\n\n    if component_spec.metadata:\n        for key, value in (component_spec.metadata.annotations or {}).items():\n            task.add_pod_annotation(key, value)\n        for key, value in (component_spec.metadata.labels or {}).items():\n            task.add_pod_label(key, value)\n\n    return task\n", "path": "sdk/python/kfp/dsl/_component_bridge.py"}], "after_files": [{"content": "# Copyright 2018 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\nimport copy\nfrom typing import Any, Mapping\nfrom ..components.structures import ComponentSpec, ComponentReference\nfrom ..components._components import _default_component_name, _resolve_command_line_and_paths\nfrom ..components._naming import _sanitize_python_function_name, generate_unique_name_conversion_table\nfrom .. import dsl\n\n\ndef _create_container_op_from_component_and_arguments(\n    component_spec: ComponentSpec,\n    arguments: Mapping[str, Any],\n    component_ref: ComponentReference = None,\n) -> 'dsl.ContainerOp':\n    # Check types of the reference arguments and serialize PipelineParams\n    arguments = arguments.copy()\n    for input_name, argument_value in arguments.items():\n        if isinstance(argument_value, dsl.PipelineParam):\n            input_type = component_spec._inputs_dict[input_name].type\n            reference_type = argument_value.param_type\n            dsl.types.verify_type_compatibility(reference_type, input_type, 'Incompatible argument passed to the input \"{}\" of component \"{}\": '.format(input_name, component_spec.name))\n\n            arguments[input_name] = str(argument_value)\n\n    resolved_cmd = _resolve_command_line_and_paths(\n        component_spec=component_spec,\n        arguments=arguments,\n    )\n\n    container_spec = component_spec.implementation.container\n\n    task = dsl.ContainerOp(\n        name=component_spec.name or _default_component_name,\n        image=container_spec.image,\n        command=resolved_cmd.command,\n        arguments=resolved_cmd.args,\n        file_outputs=resolved_cmd.output_paths,\n        artifact_argument_paths=[\n            dsl.InputArgumentPath(\n                argument=arguments[input_name],\n                input=input_name,\n                path=path,\n            )\n            for input_name, path in resolved_cmd.input_paths.items()\n        ],\n    )\n\n    component_meta = copy.copy(component_spec)\n    task._set_metadata(component_meta)\n    component_ref_without_spec = copy.copy(component_ref)\n    component_ref_without_spec.spec = None\n    task._component_ref = component_ref_without_spec\n\n    # Previously, ContainerOp had strict requirements for the output names, so we had to\n    # convert all the names before passing them to the ContainerOp constructor.\n    # Outputs with non-pythonic names could not be accessed using their original names.\n    # Now ContainerOp supports any output names, so we're now using the original output names.\n    # However to support legacy pipelines, we're also adding output references with pythonic names.\n    # TODO: Add warning when people use the legacy output names.\n    output_names = [output_spec.name for output_spec in component_spec.outputs or []] # Stabilizing the ordering\n    output_name_to_python = generate_unique_name_conversion_table(output_names, _sanitize_python_function_name)\n    for output_name in output_names:\n        pythonic_output_name = output_name_to_python[output_name]\n        # Note: Some component outputs are currently missing from task.outputs (e.g. MLPipeline UI Metadata)\n        if pythonic_output_name not in task.outputs and output_name in task.outputs:\n            task.outputs[pythonic_output_name] = task.outputs[output_name]\n\n    if container_spec.env:\n        from kubernetes import client as k8s_client\n        for name, value in container_spec.env.items():\n            task.container.add_env_variable(k8s_client.V1EnvVar(name=name, value=value))\n\n    if component_spec.metadata:\n        annotations = component_spec.metadata.annotations or {}\n        for key, value in annotations.items():\n            task.add_pod_annotation(key, value)\n        for key, value in (component_spec.metadata.labels or {}).items():\n            task.add_pod_label(key, value)\n        # Disabling the caching for the volatile components by default\n        if annotations.get('volatile_component', 'false') == 'true':\n            task.execution_options.caching_strategy.max_cache_staleness = 'P0D'\n\n    return task\n", "path": "sdk/python/kfp/dsl/_component_bridge.py"}]}