We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[Component] GCP dataproc create_cluster component cannot correctly specify image_version.
Issue:
When specifying not-null image version, create_cluster component raises:
`<HttpError 400 when requesting 
https://dataproc.googleapis.com/v1/projects/ml-pipeline-dogfood/regions/us-central1/clusters?alt=json&requestId=7c933fdacb068cd6811fb40b8334a3d4
 returned "Invalid JSON payload received. Unknown name "softwareConfig" at 'cluster': Cannot find field.">`

Initial investigation shows that in [here](https://github.com/kubeflow/pipelines/blob/7dab30085e2edda6fb4ecb61a61c9f37664009a1/components/gcp/container/component_sdk/python/kfp_component/google/dataproc/_create_cluster.py#L70), `softwareConfig` was specified as a top-layer member of cluster payload, but actually it should be a member of cluster['config']. See [this reference](https://cloud.google.com/dataproc/docs/reference/rest/v1/projects.regions.clusters#Cluster).

Will work out a fix shortly.

--- 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/gcp/container/component_sdk/python/kfp_component/google/dataproc/_create_cluster.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 import json
15 
16 from fire import decorators
17 from ._client import DataprocClient
18 from kfp_component.core import KfpExecutionContext, display
19 from .. import common as gcp_common
20 
21 @decorators.SetParseFns(image_version=str)
22 def create_cluster(project_id, region, name=None, name_prefix=None,
23     initialization_actions=None, config_bucket=None, image_version=None,
24     cluster=None, wait_interval=30):
25     """Creates a DataProc cluster under a project.
26 
27     Args:
28         project_id (str): Required. The ID of the Google Cloud Platform project 
29             that the cluster belongs to.
30         region (str): Required. The Cloud Dataproc region in which to handle the 
31             request.
32         name (str): Optional. The cluster name. Cluster names within a project
33             must be unique. Names of deleted clusters can be reused.
34         name_prefix (str): Optional. The prefix of the cluster name.
35         initialization_actions (list): Optional. List of GCS URIs of executables 
36             to execute on each node after config is completed. By default,
37             executables are run on master and all worker nodes. 
38         config_bucket (str): Optional. A Google Cloud Storage bucket used to 
39             stage job dependencies, config files, and job driver console output.
40         image_version (str): Optional. The version of software inside the cluster.
41         cluster (dict): Optional. The full cluster config. See [full details](
42             https://cloud.google.com/dataproc/docs/reference/rest/v1/projects.regions.clusters#Cluster)
43         wait_interval (int): The wait seconds between polling the operation. 
44             Defaults to 30s.
45 
46     Returns:
47         The created cluster object.
48 
49     Output Files:
50         $KFP_OUTPUT_PATH/dataproc/cluster_name.txt: The cluster name of the 
51             created cluster.
52     """
53     if not cluster:
54         cluster = {}
55     cluster['projectId'] = project_id
56     if 'config' not in cluster:
57         cluster['config'] = {}
58     if name:
59         cluster['clusterName'] = name
60     if initialization_actions:
61         cluster['config']['initializationActions'] = list(
62             map(lambda file: {
63                 'executableFile': file
64             }, initialization_actions)
65         )
66     if config_bucket:
67         cluster['config']['configBucket'] = config_bucket
68     if image_version:
69         if 'softwareConfig' not in cluster:
70             cluster['softwareConfig'] = {}
71         cluster['softwareConfig']['imageVersion'] = image_version
72 
73     return _create_cluster_internal(project_id, region, cluster, name_prefix,
74         wait_interval)
75 
76 def _create_cluster_internal(project_id, region, cluster, name_prefix, 
77     wait_interval):
78     client = DataprocClient()
79     operation_name = None
80     with KfpExecutionContext(
81         on_cancel=lambda: client.cancel_operation(operation_name)) as ctx:
82         _set_cluster_name(cluster, ctx.context_id(), name_prefix)
83         _dump_metadata(cluster, region)
84         operation = client.create_cluster(project_id, region, cluster, 
85             request_id=ctx.context_id())
86         operation_name = operation.get('name')
87         operation = client.wait_for_operation_done(operation_name, 
88             wait_interval)
89         return _dump_cluster(operation.get('response'))
90 
91 def _set_cluster_name(cluster, context_id, name_prefix):
92     if 'clusterName' in cluster:
93         return
94     if not name_prefix:
95         name_prefix = 'cluster'
96     cluster['clusterName'] = name_prefix + '-' + context_id
97 
98 def _dump_metadata(cluster, region):
99     display.display(display.Link(
100         'https://console.cloud.google.com/dataproc/clusters/{}?project={}&region={}'.format(
101             cluster.get('clusterName'), cluster.get('projectId'), region),
102         'Cluster Details'
103     ))
104 
105 def _dump_cluster(cluster):
106     gcp_common.dump_file('/tmp/kfp/output/dataproc/cluster.json', 
107         json.dumps(cluster))
108     gcp_common.dump_file('/tmp/kfp/output/dataproc/cluster_name.txt',
109         cluster.get('clusterName'))
110     return cluster
111
```
--- 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/gcp/container/component_sdk/python/kfp_component/google/dataproc/_create_cluster.py b/components/gcp/container/component_sdk/python/kfp_component/google/dataproc/_create_cluster.py
--- a/components/gcp/container/component_sdk/python/kfp_component/google/dataproc/_create_cluster.py
+++ b/components/gcp/container/component_sdk/python/kfp_component/google/dataproc/_create_cluster.py
@@ -66,9 +66,9 @@
     if config_bucket:
         cluster['config']['configBucket'] = config_bucket
     if image_version:
-        if 'softwareConfig' not in cluster:
-            cluster['softwareConfig'] = {}
-        cluster['softwareConfig']['imageVersion'] = image_version
+        if 'softwareConfig' not in cluster['config']:
+            cluster['config']['softwareConfig'] = {}
+        cluster['config']['softwareConfig']['imageVersion'] = image_version
 
     return _create_cluster_internal(project_id, region, cluster, name_prefix,
         wait_interval)

{"golden_diff": "diff --git a/components/gcp/container/component_sdk/python/kfp_component/google/dataproc/_create_cluster.py b/components/gcp/container/component_sdk/python/kfp_component/google/dataproc/_create_cluster.py\n--- a/components/gcp/container/component_sdk/python/kfp_component/google/dataproc/_create_cluster.py\n+++ b/components/gcp/container/component_sdk/python/kfp_component/google/dataproc/_create_cluster.py\n@@ -66,9 +66,9 @@\n     if config_bucket:\n         cluster['config']['configBucket'] = config_bucket\n     if image_version:\n-        if 'softwareConfig' not in cluster:\n-            cluster['softwareConfig'] = {}\n-        cluster['softwareConfig']['imageVersion'] = image_version\n+        if 'softwareConfig' not in cluster['config']:\n+            cluster['config']['softwareConfig'] = {}\n+        cluster['config']['softwareConfig']['imageVersion'] = image_version\n \n     return _create_cluster_internal(project_id, region, cluster, name_prefix,\n         wait_interval)\n", "issue": "[Component] GCP dataproc create_cluster component cannot correctly specify image_version.\nIssue:\r\nWhen specifying not-null image version, create_cluster component raises:\r\n`<HttpError 400 when requesting \r\nhttps://dataproc.googleapis.com/v1/projects/ml-pipeline-dogfood/regions/us-central1/clusters?alt=json&requestId=7c933fdacb068cd6811fb40b8334a3d4\r\n returned \"Invalid JSON payload received. Unknown name \"softwareConfig\" at 'cluster': Cannot find field.\">`\r\n\r\nInitial investigation shows that in [here](https://github.com/kubeflow/pipelines/blob/7dab30085e2edda6fb4ecb61a61c9f37664009a1/components/gcp/container/component_sdk/python/kfp_component/google/dataproc/_create_cluster.py#L70), `softwareConfig` was specified as a top-layer member of cluster payload, but actually it should be a member of cluster['config']. See [this reference](https://cloud.google.com/dataproc/docs/reference/rest/v1/projects.regions.clusters#Cluster).\r\n\r\nWill work out a fix shortly.\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.\nimport json\n\nfrom fire import decorators\nfrom ._client import DataprocClient\nfrom kfp_component.core import KfpExecutionContext, display\nfrom .. import common as gcp_common\n\n@decorators.SetParseFns(image_version=str)\ndef create_cluster(project_id, region, name=None, name_prefix=None,\n    initialization_actions=None, config_bucket=None, image_version=None,\n    cluster=None, wait_interval=30):\n    \"\"\"Creates a DataProc cluster under a project.\n\n    Args:\n        project_id (str): Required. The ID of the Google Cloud Platform project \n            that the cluster belongs to.\n        region (str): Required. The Cloud Dataproc region in which to handle the \n            request.\n        name (str): Optional. The cluster name. Cluster names within a project\n            must be unique. Names of deleted clusters can be reused.\n        name_prefix (str): Optional. The prefix of the cluster name.\n        initialization_actions (list): Optional. List of GCS URIs of executables \n            to execute on each node after config is completed. By default,\n            executables are run on master and all worker nodes. \n        config_bucket (str): Optional. A Google Cloud Storage bucket used to \n            stage job dependencies, config files, and job driver console output.\n        image_version (str): Optional. The version of software inside the cluster.\n        cluster (dict): Optional. The full cluster config. See [full details](\n            https://cloud.google.com/dataproc/docs/reference/rest/v1/projects.regions.clusters#Cluster)\n        wait_interval (int): The wait seconds between polling the operation. \n            Defaults to 30s.\n\n    Returns:\n        The created cluster object.\n\n    Output Files:\n        $KFP_OUTPUT_PATH/dataproc/cluster_name.txt: The cluster name of the \n            created cluster.\n    \"\"\"\n    if not cluster:\n        cluster = {}\n    cluster['projectId'] = project_id\n    if 'config' not in cluster:\n        cluster['config'] = {}\n    if name:\n        cluster['clusterName'] = name\n    if initialization_actions:\n        cluster['config']['initializationActions'] = list(\n            map(lambda file: {\n                'executableFile': file\n            }, initialization_actions)\n        )\n    if config_bucket:\n        cluster['config']['configBucket'] = config_bucket\n    if image_version:\n        if 'softwareConfig' not in cluster:\n            cluster['softwareConfig'] = {}\n        cluster['softwareConfig']['imageVersion'] = image_version\n\n    return _create_cluster_internal(project_id, region, cluster, name_prefix,\n        wait_interval)\n\ndef _create_cluster_internal(project_id, region, cluster, name_prefix, \n    wait_interval):\n    client = DataprocClient()\n    operation_name = None\n    with KfpExecutionContext(\n        on_cancel=lambda: client.cancel_operation(operation_name)) as ctx:\n        _set_cluster_name(cluster, ctx.context_id(), name_prefix)\n        _dump_metadata(cluster, region)\n        operation = client.create_cluster(project_id, region, cluster, \n            request_id=ctx.context_id())\n        operation_name = operation.get('name')\n        operation = client.wait_for_operation_done(operation_name, \n            wait_interval)\n        return _dump_cluster(operation.get('response'))\n\ndef _set_cluster_name(cluster, context_id, name_prefix):\n    if 'clusterName' in cluster:\n        return\n    if not name_prefix:\n        name_prefix = 'cluster'\n    cluster['clusterName'] = name_prefix + '-' + context_id\n\ndef _dump_metadata(cluster, region):\n    display.display(display.Link(\n        'https://console.cloud.google.com/dataproc/clusters/{}?project={}&region={}'.format(\n            cluster.get('clusterName'), cluster.get('projectId'), region),\n        'Cluster Details'\n    ))\n\ndef _dump_cluster(cluster):\n    gcp_common.dump_file('/tmp/kfp/output/dataproc/cluster.json', \n        json.dumps(cluster))\n    gcp_common.dump_file('/tmp/kfp/output/dataproc/cluster_name.txt',\n        cluster.get('clusterName'))\n    return cluster\n", "path": "components/gcp/container/component_sdk/python/kfp_component/google/dataproc/_create_cluster.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.\nimport json\n\nfrom fire import decorators\nfrom ._client import DataprocClient\nfrom kfp_component.core import KfpExecutionContext, display\nfrom .. import common as gcp_common\n\n@decorators.SetParseFns(image_version=str)\ndef create_cluster(project_id, region, name=None, name_prefix=None,\n    initialization_actions=None, config_bucket=None, image_version=None,\n    cluster=None, wait_interval=30):\n    \"\"\"Creates a DataProc cluster under a project.\n\n    Args:\n        project_id (str): Required. The ID of the Google Cloud Platform project \n            that the cluster belongs to.\n        region (str): Required. The Cloud Dataproc region in which to handle the \n            request.\n        name (str): Optional. The cluster name. Cluster names within a project\n            must be unique. Names of deleted clusters can be reused.\n        name_prefix (str): Optional. The prefix of the cluster name.\n        initialization_actions (list): Optional. List of GCS URIs of executables \n            to execute on each node after config is completed. By default,\n            executables are run on master and all worker nodes. \n        config_bucket (str): Optional. A Google Cloud Storage bucket used to \n            stage job dependencies, config files, and job driver console output.\n        image_version (str): Optional. The version of software inside the cluster.\n        cluster (dict): Optional. The full cluster config. See [full details](\n            https://cloud.google.com/dataproc/docs/reference/rest/v1/projects.regions.clusters#Cluster)\n        wait_interval (int): The wait seconds between polling the operation. \n            Defaults to 30s.\n\n    Returns:\n        The created cluster object.\n\n    Output Files:\n        $KFP_OUTPUT_PATH/dataproc/cluster_name.txt: The cluster name of the \n            created cluster.\n    \"\"\"\n    if not cluster:\n        cluster = {}\n    cluster['projectId'] = project_id\n    if 'config' not in cluster:\n        cluster['config'] = {}\n    if name:\n        cluster['clusterName'] = name\n    if initialization_actions:\n        cluster['config']['initializationActions'] = list(\n            map(lambda file: {\n                'executableFile': file\n            }, initialization_actions)\n        )\n    if config_bucket:\n        cluster['config']['configBucket'] = config_bucket\n    if image_version:\n        if 'softwareConfig' not in cluster['config']:\n            cluster['config']['softwareConfig'] = {}\n        cluster['config']['softwareConfig']['imageVersion'] = image_version\n\n    return _create_cluster_internal(project_id, region, cluster, name_prefix,\n        wait_interval)\n\ndef _create_cluster_internal(project_id, region, cluster, name_prefix, \n    wait_interval):\n    client = DataprocClient()\n    operation_name = None\n    with KfpExecutionContext(\n        on_cancel=lambda: client.cancel_operation(operation_name)) as ctx:\n        _set_cluster_name(cluster, ctx.context_id(), name_prefix)\n        _dump_metadata(cluster, region)\n        operation = client.create_cluster(project_id, region, cluster, \n            request_id=ctx.context_id())\n        operation_name = operation.get('name')\n        operation = client.wait_for_operation_done(operation_name, \n            wait_interval)\n        return _dump_cluster(operation.get('response'))\n\ndef _set_cluster_name(cluster, context_id, name_prefix):\n    if 'clusterName' in cluster:\n        return\n    if not name_prefix:\n        name_prefix = 'cluster'\n    cluster['clusterName'] = name_prefix + '-' + context_id\n\ndef _dump_metadata(cluster, region):\n    display.display(display.Link(\n        'https://console.cloud.google.com/dataproc/clusters/{}?project={}&region={}'.format(\n            cluster.get('clusterName'), cluster.get('projectId'), region),\n        'Cluster Details'\n    ))\n\ndef _dump_cluster(cluster):\n    gcp_common.dump_file('/tmp/kfp/output/dataproc/cluster.json', \n        json.dumps(cluster))\n    gcp_common.dump_file('/tmp/kfp/output/dataproc/cluster_name.txt',\n        cluster.get('clusterName'))\n    return cluster\n", "path": "components/gcp/container/component_sdk/python/kfp_component/google/dataproc/_create_cluster.py"}]}