prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <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.
  
  </issue>
  <code>
  [start of components/gcp/container/component_sdk/python/kfp_component/google/dataproc/_create_cluster.py]
  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 of components/gcp/container/component_sdk/python/kfp_component/google/dataproc/_create_cluster.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
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
  --- 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)
  
verification_info: {"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"}]}