We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
BrokenPipe for ml engine train component
### What steps did you take:
ML engine train component throws BrokenPipe Error after around 30 minutes of continuous training. This is not a one-off situation but jobs keep failing.
### What happened:

### What did you expect to happen:

### Environment:
<!-- Please fill in those that seem relevant. -->

How did you deploy Kubeflow Pipelines (KFP)?
Kubeflow Pipelines was deployed using gcp "ai platform pipelines" tool.
Kubeflow Pipelines 1.0.4
Kubernetes: 1.17.14-gke.1600

```
ML_ENGINE_TRAIN_OP = comp.load_component_from_url(
    "https://raw.githubusercontent.com/kubeflow/pipelines/1.3.0/components/gcp/ml_engine/train/component.yaml"
)
```
<!-- If you are not sure, here's [an introduction of all options](https://www.kubeflow.org/docs/pipelines/installation/overview/). -->

KFP version: 1.0.4

KFP SDK version: 1.3.0


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

/kind bug
<!-- Please include labels by uncommenting them to help us better triage issues, choose from the following -->
<!--
// /area frontend
// /area backend
// /area sdk
// /area testing
// /area engprod
-->
![image](https://user-images.githubusercontent.com/32985207/106754349-29749600-662d-11eb-94db-c9e30ef639fe.png)


--- 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/ml_engine/_client.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 logging
16 import time
17 
18 import googleapiclient.discovery as discovery
19 from googleapiclient import errors
20 from ..common import wait_operation_done
21 
22 class MLEngineClient:
23     """ Client for calling MLEngine APIs.
24     """
25     def __init__(self):
26         self._ml_client = discovery.build('ml', 'v1', cache_discovery=False)
27 
28     def create_job(self, project_id, job):
29         """Create a new job.
30 
31         Args:
32             project_id: the ID of the parent project.
33             job: the payload of the job.
34 
35         Returns:
36             The created job.
37         """
38         return self._ml_client.projects().jobs().create(
39             parent = 'projects/{}'.format(project_id),
40             body = job
41         ).execute()
42 
43     def cancel_job(self, project_id, job_id):
44         """Cancel the specified job.
45 
46         Args:
47             project_id: the parent project ID of the job.
48             job_id: the ID of the job.
49         """
50         job_name = 'projects/{}/jobs/{}'.format(project_id, job_id)
51         self._ml_client.projects().jobs().cancel(
52             name = job_name,
53             body = {
54                 'name': job_name
55             },
56         ).execute()
57 
58     def get_job(self, project_id, job_id):
59         """Gets the job by ID.
60 
61         Args:
62             project_id: the ID of the parent project.
63             job_id: the ID of the job to retrieve.
64         Returns:
65             The retrieved job payload.
66         """
67         job_name = 'projects/{}/jobs/{}'.format(project_id, job_id)
68         return self._ml_client.projects().jobs().get(
69             name=job_name).execute()
70 
71     def create_model(self, project_id, model):
72         """Creates a new model.
73 
74         Args:
75             project_id: the ID of the parent project.
76             model: the payload of the model.
77         Returns:
78             The created model.
79         """
80         return self._ml_client.projects().models().create(
81             parent = 'projects/{}'.format(project_id),
82             body = model
83         ).execute()
84 
85     def get_model(self, model_name):
86         """Gets a model.
87 
88         Args:
89             model_name: the name of the model.
90         Returns:
91             The retrieved model.
92         """
93         return self._ml_client.projects().models().get(
94             name = model_name
95         ).execute()
96 
97     def create_version(self, model_name, version):
98         """Creates a new version.
99 
100         Args:
101             model_name: the name of the parent model.
102             version: the payload of the version.
103 
104         Returns:
105             The created version.
106         """
107         return self._ml_client.projects().models().versions().create(
108             parent = model_name,
109             body = version
110         ).execute()
111 
112     def get_version(self, version_name):
113         """Gets a version.
114 
115         Args:
116             version_name: the name of the version.
117 
118         Returns:
119             The retrieved version. None if the version is not found.
120         """
121         try:
122             return self._ml_client.projects().models().versions().get(
123                 name = version_name
124             ).execute()
125         except errors.HttpError as e:
126             if e.resp.status == 404:
127                 return None
128             raise
129 
130     def delete_version(self, version_name):
131         """Deletes a version.
132 
133         Args:
134             version_name: the name of the version.
135 
136         Returns:
137             The delete operation. None if the version is not found.
138         """
139         try:
140             return self._ml_client.projects().models().versions().delete(
141                 name = version_name
142             ).execute()
143         except errors.HttpError as e:
144             if e.resp.status == 404:
145                 logging.info('The version has already been deleted.')
146                 return None
147             raise
148 
149     def set_default_version(self, version_name):
150         return self._ml_client.projects().models().versions().setDefault(
151             name = version_name
152         ).execute()
153 
154     def get_operation(self, operation_name):
155         """Gets an operation.
156 
157         Args:
158             operation_name: the name of the operation.
159 
160         Returns:
161             The retrieved operation.
162         """
163         return self._ml_client.projects().operations().get(
164             name = operation_name
165         ).execute()
166 
167     def wait_for_operation_done(self, operation_name, wait_interval):
168         """Waits for an operation to be done.
169 
170         Args:
171             operation_name: the name of the operation.
172             wait_interval: the wait interview between pulling job
173                 status.
174 
175         Returns:
176             The completed operation.
177         """
178         return wait_operation_done(
179             lambda: self.get_operation(operation_name), wait_interval)
180 
181     def cancel_operation(self, operation_name):
182         """Cancels an operation.
183 
184         Args:
185             operation_name: the name of the operation.
186         """
187         self._ml_client.projects().operations().cancel(
188             name = operation_name
189         ).execute()
190
```
--- 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/ml_engine/_client.py b/components/gcp/container/component_sdk/python/kfp_component/google/ml_engine/_client.py
--- a/components/gcp/container/component_sdk/python/kfp_component/google/ml_engine/_client.py
+++ b/components/gcp/container/component_sdk/python/kfp_component/google/ml_engine/_client.py
@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from functools import wraps
 import logging
 import time
 
@@ -19,12 +20,53 @@
 from googleapiclient import errors
 from ..common import wait_operation_done
 
+
+def _retry(func, tries=5, delay=1):
+    """Retry decorator for methods in MLEngineClient class.
+
+    It bypasses the BrokenPipeError by directly accessing the `_build_client` method
+    and rebuilds `_ml_client` after `delay` seconds.
+
+    Args:
+         tries (int): Total number of retries if BrokenPipeError/IOError is raised.
+         delay (int): Number of seconds to wait between consecutive retries.
+    """
+
+    @wraps(func)
+    def wrapper(self, *args, **kwargs):
+        _tries, _delay = tries, delay
+        while _tries:
+            try:
+                return func(self, *args, **kwargs)
+            except (BrokenPipeError, IOError) as e:
+                _tries -= 1
+                if not _tries:
+                    raise
+
+                logging.warning(
+                    'Caught {}. Retrying in {} seconds...'.format(
+                        e._class__.__name__, _delay
+                    )
+                )
+
+                time.sleep(_delay)
+                # access _build_client method and rebuild Http Client
+                self._build_client()
+
+    return wrapper
+
+
 class MLEngineClient:
     """ Client for calling MLEngine APIs.
     """
+
     def __init__(self):
+        self._build_client()
+
+    def _build_client(self):
         self._ml_client = discovery.build('ml', 'v1', cache_discovery=False)
 
+    @_retry
     def create_job(self, project_id, job):
         """Create a new job.
 
@@ -40,6 +82,7 @@
             body = job
         ).execute()
 
+    @_retry
     def cancel_job(self, project_id, job_id):
         """Cancel the specified job.
 
@@ -55,6 +98,7 @@
             },
         ).execute()
 
+    @_retry
     def get_job(self, project_id, job_id):
         """Gets the job by ID.
 
@@ -68,6 +112,7 @@
         return self._ml_client.projects().jobs().get(
             name=job_name).execute()
 
+    @_retry
     def create_model(self, project_id, model):
         """Creates a new model.
 
@@ -82,6 +127,7 @@
             body = model
         ).execute()
 
+    @_retry
     def get_model(self, model_name):
         """Gets a model.
 
@@ -94,6 +140,7 @@
             name = model_name
         ).execute()
 
+    @_retry
     def create_version(self, model_name, version):
         """Creates a new version.
 
@@ -109,6 +156,7 @@
             body = version
         ).execute()
 
+    @_retry
     def get_version(self, version_name):
         """Gets a version.
 
@@ -127,6 +175,7 @@
                 return None
             raise
 
+    @_retry
     def delete_version(self, version_name):
         """Deletes a version.
 
@@ -146,11 +195,13 @@
                 return None
             raise
 
+    @_retry
     def set_default_version(self, version_name):
         return self._ml_client.projects().models().versions().setDefault(
             name = version_name
         ).execute()
 
+    @_retry
     def get_operation(self, operation_name):
         """Gets an operation.
 
@@ -178,6 +229,7 @@
         return wait_operation_done(
             lambda: self.get_operation(operation_name), wait_interval)
 
+    @_retry
     def cancel_operation(self, operation_name):
         """Cancels an operation.

{"golden_diff": "diff --git a/components/gcp/container/component_sdk/python/kfp_component/google/ml_engine/_client.py b/components/gcp/container/component_sdk/python/kfp_component/google/ml_engine/_client.py\n--- a/components/gcp/container/component_sdk/python/kfp_component/google/ml_engine/_client.py\n+++ b/components/gcp/container/component_sdk/python/kfp_component/google/ml_engine/_client.py\n@@ -12,6 +12,7 @@\n # See the License for the specific language governing permissions and\n # limitations under the License.\n \n+from functools import wraps\n import logging\n import time\n \n@@ -19,12 +20,53 @@\n from googleapiclient import errors\n from ..common import wait_operation_done\n \n+\n+def _retry(func, tries=5, delay=1):\n+    \"\"\"Retry decorator for methods in MLEngineClient class.\n+\n+    It bypasses the BrokenPipeError by directly accessing the `_build_client` method\n+    and rebuilds `_ml_client` after `delay` seconds.\n+\n+    Args:\n+         tries (int): Total number of retries if BrokenPipeError/IOError is raised.\n+         delay (int): Number of seconds to wait between consecutive retries.\n+    \"\"\"\n+\n+    @wraps(func)\n+    def wrapper(self, *args, **kwargs):\n+        _tries, _delay = tries, delay\n+        while _tries:\n+            try:\n+                return func(self, *args, **kwargs)\n+            except (BrokenPipeError, IOError) as e:\n+                _tries -= 1\n+                if not _tries:\n+                    raise\n+\n+                logging.warning(\n+                    'Caught {}. Retrying in {} seconds...'.format(\n+                        e._class__.__name__, _delay\n+                    )\n+                )\n+\n+                time.sleep(_delay)\n+                # access _build_client method and rebuild Http Client\n+                self._build_client()\n+\n+    return wrapper\n+\n+\n class MLEngineClient:\n     \"\"\" Client for calling MLEngine APIs.\n     \"\"\"\n+\n     def __init__(self):\n+        self._build_client()\n+\n+    def _build_client(self):\n         self._ml_client = discovery.build('ml', 'v1', cache_discovery=False)\n \n+    @_retry\n     def create_job(self, project_id, job):\n         \"\"\"Create a new job.\n \n@@ -40,6 +82,7 @@\n             body = job\n         ).execute()\n \n+    @_retry\n     def cancel_job(self, project_id, job_id):\n         \"\"\"Cancel the specified job.\n \n@@ -55,6 +98,7 @@\n             },\n         ).execute()\n \n+    @_retry\n     def get_job(self, project_id, job_id):\n         \"\"\"Gets the job by ID.\n \n@@ -68,6 +112,7 @@\n         return self._ml_client.projects().jobs().get(\n             name=job_name).execute()\n \n+    @_retry\n     def create_model(self, project_id, model):\n         \"\"\"Creates a new model.\n \n@@ -82,6 +127,7 @@\n             body = model\n         ).execute()\n \n+    @_retry\n     def get_model(self, model_name):\n         \"\"\"Gets a model.\n \n@@ -94,6 +140,7 @@\n             name = model_name\n         ).execute()\n \n+    @_retry\n     def create_version(self, model_name, version):\n         \"\"\"Creates a new version.\n \n@@ -109,6 +156,7 @@\n             body = version\n         ).execute()\n \n+    @_retry\n     def get_version(self, version_name):\n         \"\"\"Gets a version.\n \n@@ -127,6 +175,7 @@\n                 return None\n             raise\n \n+    @_retry\n     def delete_version(self, version_name):\n         \"\"\"Deletes a version.\n \n@@ -146,11 +195,13 @@\n                 return None\n             raise\n \n+    @_retry\n     def set_default_version(self, version_name):\n         return self._ml_client.projects().models().versions().setDefault(\n             name = version_name\n         ).execute()\n \n+    @_retry\n     def get_operation(self, operation_name):\n         \"\"\"Gets an operation.\n \n@@ -178,6 +229,7 @@\n         return wait_operation_done(\n             lambda: self.get_operation(operation_name), wait_interval)\n \n+    @_retry\n     def cancel_operation(self, operation_name):\n         \"\"\"Cancels an operation.\n", "issue": "BrokenPipe for ml engine train component\n### What steps did you take:\r\nML engine train component throws BrokenPipe Error after around 30 minutes of continuous training. This is not a one-off situation but jobs keep failing.\r\n### What happened:\r\n\r\n### What did you expect to happen:\r\n\r\n### Environment:\r\n<!-- Please fill in those that seem relevant. -->\r\n\r\nHow did you deploy Kubeflow Pipelines (KFP)?\r\nKubeflow Pipelines was deployed using gcp \"ai platform pipelines\" tool.\r\nKubeflow Pipelines 1.0.4\r\nKubernetes: 1.17.14-gke.1600\r\n\r\n```\r\nML_ENGINE_TRAIN_OP = comp.load_component_from_url(\r\n    \"https://raw.githubusercontent.com/kubeflow/pipelines/1.3.0/components/gcp/ml_engine/train/component.yaml\"\r\n)\r\n```\r\n<!-- If you are not sure, here's [an introduction of all options](https://www.kubeflow.org/docs/pipelines/installation/overview/). -->\r\n\r\nKFP version: 1.0.4\r\n\r\nKFP SDK version: 1.3.0\r\n\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/kind bug\r\n<!-- Please include labels by uncommenting them to help us better triage issues, choose from the following -->\r\n<!--\r\n// /area frontend\r\n// /area backend\r\n// /area sdk\r\n// /area testing\r\n// /area engprod\r\n-->\r\n![image](https://user-images.githubusercontent.com/32985207/106754349-29749600-662d-11eb-94db-c9e30ef639fe.png)\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 logging\nimport time\n\nimport googleapiclient.discovery as discovery\nfrom googleapiclient import errors\nfrom ..common import wait_operation_done\n\nclass MLEngineClient:\n    \"\"\" Client for calling MLEngine APIs.\n    \"\"\"\n    def __init__(self):\n        self._ml_client = discovery.build('ml', 'v1', cache_discovery=False)\n\n    def create_job(self, project_id, job):\n        \"\"\"Create a new job.\n\n        Args:\n            project_id: the ID of the parent project.\n            job: the payload of the job.\n\n        Returns:\n            The created job.\n        \"\"\"\n        return self._ml_client.projects().jobs().create(\n            parent = 'projects/{}'.format(project_id),\n            body = job\n        ).execute()\n\n    def cancel_job(self, project_id, job_id):\n        \"\"\"Cancel the specified job.\n\n        Args:\n            project_id: the parent project ID of the job.\n            job_id: the ID of the job.\n        \"\"\"\n        job_name = 'projects/{}/jobs/{}'.format(project_id, job_id)\n        self._ml_client.projects().jobs().cancel(\n            name = job_name,\n            body = {\n                'name': job_name\n            },\n        ).execute()\n\n    def get_job(self, project_id, job_id):\n        \"\"\"Gets the job by ID.\n\n        Args:\n            project_id: the ID of the parent project.\n            job_id: the ID of the job to retrieve.\n        Returns:\n            The retrieved job payload.\n        \"\"\"\n        job_name = 'projects/{}/jobs/{}'.format(project_id, job_id)\n        return self._ml_client.projects().jobs().get(\n            name=job_name).execute()\n\n    def create_model(self, project_id, model):\n        \"\"\"Creates a new model.\n\n        Args:\n            project_id: the ID of the parent project.\n            model: the payload of the model.\n        Returns:\n            The created model.\n        \"\"\"\n        return self._ml_client.projects().models().create(\n            parent = 'projects/{}'.format(project_id),\n            body = model\n        ).execute()\n\n    def get_model(self, model_name):\n        \"\"\"Gets a model.\n\n        Args:\n            model_name: the name of the model.\n        Returns:\n            The retrieved model.\n        \"\"\"\n        return self._ml_client.projects().models().get(\n            name = model_name\n        ).execute()\n\n    def create_version(self, model_name, version):\n        \"\"\"Creates a new version.\n\n        Args:\n            model_name: the name of the parent model.\n            version: the payload of the version.\n\n        Returns:\n            The created version.\n        \"\"\"\n        return self._ml_client.projects().models().versions().create(\n            parent = model_name,\n            body = version\n        ).execute()\n\n    def get_version(self, version_name):\n        \"\"\"Gets a version.\n\n        Args:\n            version_name: the name of the version.\n\n        Returns:\n            The retrieved version. None if the version is not found.\n        \"\"\"\n        try:\n            return self._ml_client.projects().models().versions().get(\n                name = version_name\n            ).execute()\n        except errors.HttpError as e:\n            if e.resp.status == 404:\n                return None\n            raise\n\n    def delete_version(self, version_name):\n        \"\"\"Deletes a version.\n\n        Args:\n            version_name: the name of the version.\n\n        Returns:\n            The delete operation. None if the version is not found.\n        \"\"\"\n        try:\n            return self._ml_client.projects().models().versions().delete(\n                name = version_name\n            ).execute()\n        except errors.HttpError as e:\n            if e.resp.status == 404:\n                logging.info('The version has already been deleted.')\n                return None\n            raise\n\n    def set_default_version(self, version_name):\n        return self._ml_client.projects().models().versions().setDefault(\n            name = version_name\n        ).execute()\n\n    def get_operation(self, operation_name):\n        \"\"\"Gets an operation.\n\n        Args:\n            operation_name: the name of the operation.\n\n        Returns:\n            The retrieved operation.\n        \"\"\"\n        return self._ml_client.projects().operations().get(\n            name = operation_name\n        ).execute()\n\n    def wait_for_operation_done(self, operation_name, wait_interval):\n        \"\"\"Waits for an operation to be done.\n\n        Args:\n            operation_name: the name of the operation.\n            wait_interval: the wait interview between pulling job\n                status.\n\n        Returns:\n            The completed operation.\n        \"\"\"\n        return wait_operation_done(\n            lambda: self.get_operation(operation_name), wait_interval)\n\n    def cancel_operation(self, operation_name):\n        \"\"\"Cancels an operation.\n\n        Args:\n            operation_name: the name of the operation.\n        \"\"\"\n        self._ml_client.projects().operations().cancel(\n            name = operation_name\n        ).execute()\n", "path": "components/gcp/container/component_sdk/python/kfp_component/google/ml_engine/_client.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\nfrom functools import wraps\nimport logging\nimport time\n\nimport googleapiclient.discovery as discovery\nfrom googleapiclient import errors\nfrom ..common import wait_operation_done\n\n\ndef _retry(func, tries=5, delay=1):\n    \"\"\"Retry decorator for methods in MLEngineClient class.\n\n    It bypasses the BrokenPipeError by directly accessing the `_build_client` method\n    and rebuilds `_ml_client` after `delay` seconds.\n\n    Args:\n         tries (int): Total number of retries if BrokenPipeError/IOError is raised.\n         delay (int): Number of seconds to wait between consecutive retries.\n    \"\"\"\n\n    @wraps(func)\n    def wrapper(self, *args, **kwargs):\n        _tries, _delay = tries, delay\n        while _tries:\n            try:\n                return func(self, *args, **kwargs)\n            except (BrokenPipeError, IOError) as e:\n                _tries -= 1\n                if not _tries:\n                    raise\n\n                logging.warning(\n                    'Caught {}. Retrying in {} seconds...'.format(\n                        e._class__.__name__, _delay\n                    )\n                )\n\n                time.sleep(_delay)\n                # access _build_client method and rebuild Http Client\n                self._build_client()\n\n    return wrapper\n\n\nclass MLEngineClient:\n    \"\"\" Client for calling MLEngine APIs.\n    \"\"\"\n\n    def __init__(self):\n        self._build_client()\n\n    def _build_client(self):\n        self._ml_client = discovery.build('ml', 'v1', cache_discovery=False)\n\n    @_retry\n    def create_job(self, project_id, job):\n        \"\"\"Create a new job.\n\n        Args:\n            project_id: the ID of the parent project.\n            job: the payload of the job.\n\n        Returns:\n            The created job.\n        \"\"\"\n        return self._ml_client.projects().jobs().create(\n            parent = 'projects/{}'.format(project_id),\n            body = job\n        ).execute()\n\n    @_retry\n    def cancel_job(self, project_id, job_id):\n        \"\"\"Cancel the specified job.\n\n        Args:\n            project_id: the parent project ID of the job.\n            job_id: the ID of the job.\n        \"\"\"\n        job_name = 'projects/{}/jobs/{}'.format(project_id, job_id)\n        self._ml_client.projects().jobs().cancel(\n            name = job_name,\n            body = {\n                'name': job_name\n            },\n        ).execute()\n\n    @_retry\n    def get_job(self, project_id, job_id):\n        \"\"\"Gets the job by ID.\n\n        Args:\n            project_id: the ID of the parent project.\n            job_id: the ID of the job to retrieve.\n        Returns:\n            The retrieved job payload.\n        \"\"\"\n        job_name = 'projects/{}/jobs/{}'.format(project_id, job_id)\n        return self._ml_client.projects().jobs().get(\n            name=job_name).execute()\n\n    @_retry\n    def create_model(self, project_id, model):\n        \"\"\"Creates a new model.\n\n        Args:\n            project_id: the ID of the parent project.\n            model: the payload of the model.\n        Returns:\n            The created model.\n        \"\"\"\n        return self._ml_client.projects().models().create(\n            parent = 'projects/{}'.format(project_id),\n            body = model\n        ).execute()\n\n    @_retry\n    def get_model(self, model_name):\n        \"\"\"Gets a model.\n\n        Args:\n            model_name: the name of the model.\n        Returns:\n            The retrieved model.\n        \"\"\"\n        return self._ml_client.projects().models().get(\n            name = model_name\n        ).execute()\n\n    @_retry\n    def create_version(self, model_name, version):\n        \"\"\"Creates a new version.\n\n        Args:\n            model_name: the name of the parent model.\n            version: the payload of the version.\n\n        Returns:\n            The created version.\n        \"\"\"\n        return self._ml_client.projects().models().versions().create(\n            parent = model_name,\n            body = version\n        ).execute()\n\n    @_retry\n    def get_version(self, version_name):\n        \"\"\"Gets a version.\n\n        Args:\n            version_name: the name of the version.\n\n        Returns:\n            The retrieved version. None if the version is not found.\n        \"\"\"\n        try:\n            return self._ml_client.projects().models().versions().get(\n                name = version_name\n            ).execute()\n        except errors.HttpError as e:\n            if e.resp.status == 404:\n                return None\n            raise\n\n    @_retry\n    def delete_version(self, version_name):\n        \"\"\"Deletes a version.\n\n        Args:\n            version_name: the name of the version.\n\n        Returns:\n            The delete operation. None if the version is not found.\n        \"\"\"\n        try:\n            return self._ml_client.projects().models().versions().delete(\n                name = version_name\n            ).execute()\n        except errors.HttpError as e:\n            if e.resp.status == 404:\n                logging.info('The version has already been deleted.')\n                return None\n            raise\n\n    @_retry\n    def set_default_version(self, version_name):\n        return self._ml_client.projects().models().versions().setDefault(\n            name = version_name\n        ).execute()\n\n    @_retry\n    def get_operation(self, operation_name):\n        \"\"\"Gets an operation.\n\n        Args:\n            operation_name: the name of the operation.\n\n        Returns:\n            The retrieved operation.\n        \"\"\"\n        return self._ml_client.projects().operations().get(\n            name = operation_name\n        ).execute()\n\n    def wait_for_operation_done(self, operation_name, wait_interval):\n        \"\"\"Waits for an operation to be done.\n\n        Args:\n            operation_name: the name of the operation.\n            wait_interval: the wait interview between pulling job\n                status.\n\n        Returns:\n            The completed operation.\n        \"\"\"\n        return wait_operation_done(\n            lambda: self.get_operation(operation_name), wait_interval)\n\n    @_retry\n    def cancel_operation(self, operation_name):\n        \"\"\"Cancels an operation.\n\n        Args:\n            operation_name: the name of the operation.\n        \"\"\"\n        self._ml_client.projects().operations().cancel(\n            name = operation_name\n        ).execute()\n", "path": "components/gcp/container/component_sdk/python/kfp_component/google/ml_engine/_client.py"}]}