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: `components/aws/emr/common/_utils.py`
Content:
```
1 # Licensed under the Apache License, Version 2.0 (the "License");
2 # you may not use this file except in compliance with the License.
3 # You may obtain a copy of the License at
4 #
5 # http://www.apache.org/licenses/LICENSE-2.0
6 #
7 # Unless required by applicable law or agreed to in writing, software
8 # distributed under the License is distributed on an "AS IS" BASIS,
9 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10 # See the License for the specific language governing permissions and
11 # limitations under the License.
12 
13 
14 import datetime
15 import os
16 import subprocess
17 import time
18 
19 import boto3
20 from botocore.exceptions import ClientError
21 import json
22 
23 
24 def get_client(region=None):
25     """Builds a client to the AWS EMR API."""
26     client = boto3.client('emr', region_name=region)
27     return client
28 
29 def create_cluster(client, cluster_name, log_s3_uri, release_label, instance_type, instance_count):
30   """Create a EMR cluster."""
31   response = client.run_job_flow(
32     Name=cluster_name,
33     LogUri=log_s3_uri,
34     ReleaseLabel=release_label,
35     Applications=[
36       {
37         'Name': 'Spark'
38       }
39     ],
40     BootstrapActions=[
41       {
42         'Name': 'Maximize Spark Default Config',
43         'ScriptBootstrapAction': {
44           'Path': 's3://support.elasticmapreduce/spark/maximize-spark-default-config',
45         }
46       },
47     ],
48     Instances= {
49         'MasterInstanceType': instance_type,
50         'SlaveInstanceType': instance_type,
51         'InstanceCount': instance_count,
52         'KeepJobFlowAliveWhenNoSteps':True,
53         'TerminationProtected':False,
54 
55     },
56     VisibleToAllUsers=True,
57     JobFlowRole='EMR_EC2_DefaultRole',
58     ServiceRole='EMR_DefaultRole'
59   )
60   return response
61 
62 
63 def delete_cluster(client, jobflow_id):
64   """Delete a EMR cluster. Cluster shutdowns in background"""
65   client.terminate_job_flows(JobFlowIds=[jobflow_id])
66 
67 def wait_for_cluster(client, jobflow_id):
68   """Waiting for a new cluster to be ready."""
69   while True:
70     response = client.describe_cluster(ClusterId=jobflow_id)
71     cluster_status = response['Cluster']['Status']
72     state = cluster_status['State']
73 
74     if 'Message' in cluster_status['StateChangeReason']:
75       state = cluster_status['State']
76       message = cluster_status['StateChangeReason']['Message']
77 
78       if state in ['TERMINATED', 'TERMINATED', 'TERMINATED_WITH_ERRORS']:
79         raise Exception(message)
80 
81       if state == 'WAITING':
82         print('EMR cluster create completed')
83         break
84 
85     print("Cluster state: {}, wait 15s for cluster to start up.".format(state))
86     time.sleep(15)
87 
88 # Check following documentation to add other job type steps. Seems python SDK only have 'HadoopJarStep' here.
89 # https://docs.aws.amazon.com/cli/latest/reference/emr/add-steps.html
90 def submit_spark_job(client, jobflow_id, job_name, jar_path, main_class, extra_args):
91   """Submits single spark job to a running cluster"""
92 
93   spark_job = {
94     'Name': job_name,
95     'ActionOnFailure': 'CONTINUE',
96     'HadoopJarStep': {
97       'Jar': 'command-runner.jar'
98     }
99   }
100 
101   spark_args = ['spark-submit', "--deploy-mode", "cluster"]
102   if main_class:
103     spark_args.extend(['--class', main_class])
104   spark_args.extend([jar_path])
105   spark_args.extend(extra_args)
106 
107   spark_job['HadoopJarStep']['Args'] = spark_args
108 
109   try:
110     response = client.add_job_flow_steps(
111             JobFlowId=jobflow_id,
112             Steps=[spark_job],
113     )
114   except ClientError as e:
115     print(e.response['Error']['Message'])
116     exit(1)
117 
118   step_id = response['StepIds'][0]
119   print("Step Id {} has been submitted".format(step_id))
120   return step_id
121 
122 def wait_for_job(client, jobflow_id, step_id):
123   """Waiting for a cluster step by polling it."""
124   while True:
125     result = client.describe_step(ClusterId=jobflow_id, StepId=step_id)
126     step_status = result['Step']['Status']
127     state = step_status['State']
128 
129     if state in ('CANCELLED', 'FAILED', 'INTERRUPTED'):
130       err_msg = 'UNKNOWN'
131       if 'FailureDetails' in step_status:
132         err_msg = step_status['FailureDetails']
133 
134       raise Exception(err_msg)
135     elif state == 'COMPLETED':
136       print('EMR Step finishes')
137       break
138 
139     print("Step state: {}, wait 15s for step status update.".format(state))
140     time.sleep(10)
141 
142 def submit_pyspark_job(client, jobflow_id, job_name, py_file, extra_args):
143   """Submits single spark job to a running cluster"""
144 
145   pyspark_args = ['spark-submit', py_file]
146   pyspark_args.extend(extra_args)
147   return submit_spark_job(client, jobflow_id, job_name, 'command-runner.jar', '', pyspark_args)
148
```
--- 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/components/aws/emr/common/_utils.py b/components/aws/emr/common/_utils.py
--- a/components/aws/emr/common/_utils.py
+++ b/components/aws/emr/common/_utils.py
@@ -26,8 +26,23 @@
     client = boto3.client('emr', region_name=region)
     return client
 
-def create_cluster(client, cluster_name, log_s3_uri, release_label, instance_type, instance_count):
+def create_cluster(client, cluster_name, log_s3_uri, release_label, instance_type, instance_count, ec2SubnetId=None, ec2KeyName=None):
   """Create a EMR cluster."""
+
+  instances = {
+      'MasterInstanceType': instance_type,
+      'SlaveInstanceType': instance_type,
+      'InstanceCount': instance_count,
+      'KeepJobFlowAliveWhenNoSteps':True,
+      'TerminationProtected':False
+  }
+
+  if ec2SubnetId is not None:
+    instances['Ec2SubnetId'] = ec2SubnetId
+
+  if ec2KeyName is not None:
+    instances['Ec2KeyName'] = ec2KeyName
+
   response = client.run_job_flow(
     Name=cluster_name,
     LogUri=log_s3_uri,
@@ -45,14 +60,7 @@
         }
       },
     ],
-    Instances= {
-        'MasterInstanceType': instance_type,
-        'SlaveInstanceType': instance_type,
-        'InstanceCount': instance_count,
-        'KeepJobFlowAliveWhenNoSteps':True,
-        'TerminationProtected':False,
-
-    },
+    Instances= instances,
     VisibleToAllUsers=True,
     JobFlowRole='EMR_EC2_DefaultRole',
     ServiceRole='EMR_DefaultRole'
@@ -142,6 +150,5 @@
 def submit_pyspark_job(client, jobflow_id, job_name, py_file, extra_args):
   """Submits single spark job to a running cluster"""
 
-  pyspark_args = ['spark-submit', py_file]
-  pyspark_args.extend(extra_args)
-  return submit_spark_job(client, jobflow_id, job_name, 'command-runner.jar', '', pyspark_args)
+  pyspark_args = [py_file, extra_args]
+  return submit_spark_job(client, jobflow_id, job_name, '', '', pyspark_args)

{"golden_diff": "diff --git a/components/aws/emr/common/_utils.py b/components/aws/emr/common/_utils.py\n--- a/components/aws/emr/common/_utils.py\n+++ b/components/aws/emr/common/_utils.py\n@@ -26,8 +26,23 @@\n     client = boto3.client('emr', region_name=region)\n     return client\n \n-def create_cluster(client, cluster_name, log_s3_uri, release_label, instance_type, instance_count):\n+def create_cluster(client, cluster_name, log_s3_uri, release_label, instance_type, instance_count, ec2SubnetId=None, ec2KeyName=None):\n   \"\"\"Create a EMR cluster.\"\"\"\n+\n+  instances = {\n+      'MasterInstanceType': instance_type,\n+      'SlaveInstanceType': instance_type,\n+      'InstanceCount': instance_count,\n+      'KeepJobFlowAliveWhenNoSteps':True,\n+      'TerminationProtected':False\n+  }\n+\n+  if ec2SubnetId is not None:\n+    instances['Ec2SubnetId'] = ec2SubnetId\n+\n+  if ec2KeyName is not None:\n+    instances['Ec2KeyName'] = ec2KeyName\n+\n   response = client.run_job_flow(\n     Name=cluster_name,\n     LogUri=log_s3_uri,\n@@ -45,14 +60,7 @@\n         }\n       },\n     ],\n-    Instances= {\n-        'MasterInstanceType': instance_type,\n-        'SlaveInstanceType': instance_type,\n-        'InstanceCount': instance_count,\n-        'KeepJobFlowAliveWhenNoSteps':True,\n-        'TerminationProtected':False,\n-\n-    },\n+    Instances= instances,\n     VisibleToAllUsers=True,\n     JobFlowRole='EMR_EC2_DefaultRole',\n     ServiceRole='EMR_DefaultRole'\n@@ -142,6 +150,5 @@\n def submit_pyspark_job(client, jobflow_id, job_name, py_file, extra_args):\n   \"\"\"Submits single spark job to a running cluster\"\"\"\n \n-  pyspark_args = ['spark-submit', py_file]\n-  pyspark_args.extend(extra_args)\n-  return submit_spark_job(client, jobflow_id, job_name, 'command-runner.jar', '', pyspark_args)\n+  pyspark_args = [py_file, extra_args]\n+  return submit_spark_job(client, jobflow_id, job_name, '', '', pyspark_args)\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": "# 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 datetime\nimport os\nimport subprocess\nimport time\n\nimport boto3\nfrom botocore.exceptions import ClientError\nimport json\n\n\ndef get_client(region=None):\n    \"\"\"Builds a client to the AWS EMR API.\"\"\"\n    client = boto3.client('emr', region_name=region)\n    return client\n\ndef create_cluster(client, cluster_name, log_s3_uri, release_label, instance_type, instance_count):\n  \"\"\"Create a EMR cluster.\"\"\"\n  response = client.run_job_flow(\n    Name=cluster_name,\n    LogUri=log_s3_uri,\n    ReleaseLabel=release_label,\n    Applications=[\n      {\n        'Name': 'Spark'\n      }\n    ],\n    BootstrapActions=[\n      {\n        'Name': 'Maximize Spark Default Config',\n        'ScriptBootstrapAction': {\n          'Path': 's3://support.elasticmapreduce/spark/maximize-spark-default-config',\n        }\n      },\n    ],\n    Instances= {\n        'MasterInstanceType': instance_type,\n        'SlaveInstanceType': instance_type,\n        'InstanceCount': instance_count,\n        'KeepJobFlowAliveWhenNoSteps':True,\n        'TerminationProtected':False,\n\n    },\n    VisibleToAllUsers=True,\n    JobFlowRole='EMR_EC2_DefaultRole',\n    ServiceRole='EMR_DefaultRole'\n  )\n  return response\n\n\ndef delete_cluster(client, jobflow_id):\n  \"\"\"Delete a EMR cluster. Cluster shutdowns in background\"\"\"\n  client.terminate_job_flows(JobFlowIds=[jobflow_id])\n\ndef wait_for_cluster(client, jobflow_id):\n  \"\"\"Waiting for a new cluster to be ready.\"\"\"\n  while True:\n    response = client.describe_cluster(ClusterId=jobflow_id)\n    cluster_status = response['Cluster']['Status']\n    state = cluster_status['State']\n\n    if 'Message' in cluster_status['StateChangeReason']:\n      state = cluster_status['State']\n      message = cluster_status['StateChangeReason']['Message']\n\n      if state in ['TERMINATED', 'TERMINATED', 'TERMINATED_WITH_ERRORS']:\n        raise Exception(message)\n\n      if state == 'WAITING':\n        print('EMR cluster create completed')\n        break\n\n    print(\"Cluster state: {}, wait 15s for cluster to start up.\".format(state))\n    time.sleep(15)\n\n# Check following documentation to add other job type steps. Seems python SDK only have 'HadoopJarStep' here.\n# https://docs.aws.amazon.com/cli/latest/reference/emr/add-steps.html\ndef submit_spark_job(client, jobflow_id, job_name, jar_path, main_class, extra_args):\n  \"\"\"Submits single spark job to a running cluster\"\"\"\n\n  spark_job = {\n    'Name': job_name,\n    'ActionOnFailure': 'CONTINUE',\n    'HadoopJarStep': {\n      'Jar': 'command-runner.jar'\n    }\n  }\n\n  spark_args = ['spark-submit', \"--deploy-mode\", \"cluster\"]\n  if main_class:\n    spark_args.extend(['--class', main_class])\n  spark_args.extend([jar_path])\n  spark_args.extend(extra_args)\n\n  spark_job['HadoopJarStep']['Args'] = spark_args\n\n  try:\n    response = client.add_job_flow_steps(\n            JobFlowId=jobflow_id,\n            Steps=[spark_job],\n    )\n  except ClientError as e:\n    print(e.response['Error']['Message'])\n    exit(1)\n\n  step_id = response['StepIds'][0]\n  print(\"Step Id {} has been submitted\".format(step_id))\n  return step_id\n\ndef wait_for_job(client, jobflow_id, step_id):\n  \"\"\"Waiting for a cluster step by polling it.\"\"\"\n  while True:\n    result = client.describe_step(ClusterId=jobflow_id, StepId=step_id)\n    step_status = result['Step']['Status']\n    state = step_status['State']\n\n    if state in ('CANCELLED', 'FAILED', 'INTERRUPTED'):\n      err_msg = 'UNKNOWN'\n      if 'FailureDetails' in step_status:\n        err_msg = step_status['FailureDetails']\n\n      raise Exception(err_msg)\n    elif state == 'COMPLETED':\n      print('EMR Step finishes')\n      break\n\n    print(\"Step state: {}, wait 15s for step status update.\".format(state))\n    time.sleep(10)\n\ndef submit_pyspark_job(client, jobflow_id, job_name, py_file, extra_args):\n  \"\"\"Submits single spark job to a running cluster\"\"\"\n\n  pyspark_args = ['spark-submit', py_file]\n  pyspark_args.extend(extra_args)\n  return submit_spark_job(client, jobflow_id, job_name, 'command-runner.jar', '', pyspark_args)\n", "path": "components/aws/emr/common/_utils.py"}], "after_files": [{"content": "# 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 datetime\nimport os\nimport subprocess\nimport time\n\nimport boto3\nfrom botocore.exceptions import ClientError\nimport json\n\n\ndef get_client(region=None):\n    \"\"\"Builds a client to the AWS EMR API.\"\"\"\n    client = boto3.client('emr', region_name=region)\n    return client\n\ndef create_cluster(client, cluster_name, log_s3_uri, release_label, instance_type, instance_count, ec2SubnetId=None, ec2KeyName=None):\n  \"\"\"Create a EMR cluster.\"\"\"\n\n  instances = {\n      'MasterInstanceType': instance_type,\n      'SlaveInstanceType': instance_type,\n      'InstanceCount': instance_count,\n      'KeepJobFlowAliveWhenNoSteps':True,\n      'TerminationProtected':False\n  }\n\n  if ec2SubnetId is not None:\n    instances['Ec2SubnetId'] = ec2SubnetId\n\n  if ec2KeyName is not None:\n    instances['Ec2KeyName'] = ec2KeyName\n\n  response = client.run_job_flow(\n    Name=cluster_name,\n    LogUri=log_s3_uri,\n    ReleaseLabel=release_label,\n    Applications=[\n      {\n        'Name': 'Spark'\n      }\n    ],\n    BootstrapActions=[\n      {\n        'Name': 'Maximize Spark Default Config',\n        'ScriptBootstrapAction': {\n          'Path': 's3://support.elasticmapreduce/spark/maximize-spark-default-config',\n        }\n      },\n    ],\n    Instances= instances,\n    VisibleToAllUsers=True,\n    JobFlowRole='EMR_EC2_DefaultRole',\n    ServiceRole='EMR_DefaultRole'\n  )\n  return response\n\n\ndef delete_cluster(client, jobflow_id):\n  \"\"\"Delete a EMR cluster. Cluster shutdowns in background\"\"\"\n  client.terminate_job_flows(JobFlowIds=[jobflow_id])\n\ndef wait_for_cluster(client, jobflow_id):\n  \"\"\"Waiting for a new cluster to be ready.\"\"\"\n  while True:\n    response = client.describe_cluster(ClusterId=jobflow_id)\n    cluster_status = response['Cluster']['Status']\n    state = cluster_status['State']\n\n    if 'Message' in cluster_status['StateChangeReason']:\n      state = cluster_status['State']\n      message = cluster_status['StateChangeReason']['Message']\n\n      if state in ['TERMINATED', 'TERMINATED', 'TERMINATED_WITH_ERRORS']:\n        raise Exception(message)\n\n      if state == 'WAITING':\n        print('EMR cluster create completed')\n        break\n\n    print(\"Cluster state: {}, wait 15s for cluster to start up.\".format(state))\n    time.sleep(15)\n\n# Check following documentation to add other job type steps. Seems python SDK only have 'HadoopJarStep' here.\n# https://docs.aws.amazon.com/cli/latest/reference/emr/add-steps.html\ndef submit_spark_job(client, jobflow_id, job_name, jar_path, main_class, extra_args):\n  \"\"\"Submits single spark job to a running cluster\"\"\"\n\n  spark_job = {\n    'Name': job_name,\n    'ActionOnFailure': 'CONTINUE',\n    'HadoopJarStep': {\n      'Jar': 'command-runner.jar'\n    }\n  }\n\n  spark_args = ['spark-submit', \"--deploy-mode\", \"cluster\"]\n  if main_class:\n    spark_args.extend(['--class', main_class])\n  spark_args.extend([jar_path])\n  spark_args.extend(extra_args)\n\n  spark_job['HadoopJarStep']['Args'] = spark_args\n\n  try:\n    response = client.add_job_flow_steps(\n            JobFlowId=jobflow_id,\n            Steps=[spark_job],\n    )\n  except ClientError as e:\n    print(e.response['Error']['Message'])\n    exit(1)\n\n  step_id = response['StepIds'][0]\n  print(\"Step Id {} has been submitted\".format(step_id))\n  return step_id\n\ndef wait_for_job(client, jobflow_id, step_id):\n  \"\"\"Waiting for a cluster step by polling it.\"\"\"\n  while True:\n    result = client.describe_step(ClusterId=jobflow_id, StepId=step_id)\n    step_status = result['Step']['Status']\n    state = step_status['State']\n\n    if state in ('CANCELLED', 'FAILED', 'INTERRUPTED'):\n      err_msg = 'UNKNOWN'\n      if 'FailureDetails' in step_status:\n        err_msg = step_status['FailureDetails']\n\n      raise Exception(err_msg)\n    elif state == 'COMPLETED':\n      print('EMR Step finishes')\n      break\n\n    print(\"Step state: {}, wait 15s for step status update.\".format(state))\n    time.sleep(10)\n\ndef submit_pyspark_job(client, jobflow_id, job_name, py_file, extra_args):\n  \"\"\"Submits single spark job to a running cluster\"\"\"\n\n  pyspark_args = [py_file, extra_args]\n  return submit_spark_job(client, jobflow_id, job_name, '', '', pyspark_args)\n", "path": "components/aws/emr/common/_utils.py"}]}