We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
KubernetesExecutor leaves failed pods due to deepcopy issue with Google providers
### Apache Airflow version

Other Airflow 2 version (please specify below)

### What happened

With Airflow 2.3 and 2.4 there appears to be a bug in the KubernetesExecutor when used in conjunction with the Google airflow providers. This bug does not affect Airflow 2.2 due to the pip version requirements.

The bug specifically presents itself when using nearly any Google provider operator. During the pod lifecycle, all is well until the executor in the pod starts to clean up following a successful run. Airflow itself still see's the task marked as a success, but in Kubernetes, while the task is finishing up after reporting status, it actually crashes and puts the pod into a Failed state silently:
```
Traceback (most recent call last):
  File "/home/airflow/.local/bin/airflow", line 8, in <module>
    sys.exit(main())
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/__main__.py", line 39, in main
    args.func(args)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/cli_parser.py", line 52, in command
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/cli.py", line 103, in wrapper
    return f(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/commands/task_command.py", line 382, in task_run
    _run_task_by_selected_method(args, dag, ti)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/commands/task_command.py", line 189, in _run_task_by_selected_method
    _run_task_by_local_task_job(args, ti)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/commands/task_command.py", line 247, in _run_task_by_local_task_job
    run_job.run()
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/base_job.py", line 247, in run
    self._execute()
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/local_task_job.py", line 137, in _execute
    self.handle_task_exit(return_code)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/local_task_job.py", line 168, in handle_task_exit
    self._run_mini_scheduler_on_child_tasks()
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/session.py", line 75, in wrapper
    return func(*args, session=session, **kwargs)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/local_task_job.py", line 253, in _run_mini_scheduler_on_child_tasks
    partial_dag = task.dag.partial_subset(
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/dag.py", line 2188, in partial_subset
    dag.task_dict = {
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/dag.py", line 2189, in <dictcomp>
    t.task_id: _deepcopy_task(t)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/dag.py", line 2186, in _deepcopy_task
    return copy.deepcopy(t, memo)
  File "/usr/local/lib/python3.9/copy.py", line 153, in deepcopy
    y = copier(memo)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/baseoperator.py", line 1163, in __deepcopy__
    setattr(result, k, copy.deepcopy(v, memo))
  File "/usr/local/lib/python3.9/copy.py", line 172, in deepcopy
    y = _reconstruct(x, memo, *rv)
  File "/usr/local/lib/python3.9/copy.py", line 264, in _reconstruct
    y = func(*args)
  File "/usr/local/lib/python3.9/enum.py", line 384, in __call__
    return cls.__new__(cls, value)
  File "/usr/local/lib/python3.9/enum.py", line 702, in __new__
    raise ve_exc
ValueError: <object object at 0x7f570181a3c0> is not a valid _MethodDefault
```

Based on a quick look, it appears to be related to the default argument that Google is using in its operators which happens to be an Enum, and fails during a deepcopy at the end of the task.

Example operator that is affected: https://github.com/apache/airflow/blob/403ed7163f3431deb7fc21108e1743385e139907/airflow/providers/google/cloud/hooks/dataproc.py#L753
Reference to the Google Python API core which has the Enum causing the problem: https://github.com/googleapis/python-api-core/blob/main/google/api_core/gapic_v1/method.py#L31

### What you think should happen instead

Kubernetes pods should succeed, be marked as `Completed`, and then be gracefully terminated.

### How to reproduce

Use any `apache-airflow-providers-google` >= 7.0.0 which includes `google-api-core` >= 2.2.2. Run a DAG with a task which uses any of the Google operators which have `_MethodDefault` as a default argument.

### Operating System

Debian GNU/Linux 11 (bullseye)

### Versions of Apache Airflow Providers

apache-airflow-providers-amazon==6.0.0
apache-airflow-providers-apache-hive==5.0.0
apache-airflow-providers-celery==3.0.0
apache-airflow-providers-cncf-kubernetes==4.4.0
apache-airflow-providers-common-sql==1.3.1
apache-airflow-providers-docker==3.2.0
apache-airflow-providers-elasticsearch==4.2.1
apache-airflow-providers-ftp==3.1.0
apache-airflow-providers-google==8.4.0
apache-airflow-providers-grpc==3.0.0
apache-airflow-providers-hashicorp==3.1.0
apache-airflow-providers-http==4.0.0
apache-airflow-providers-imap==3.0.0
apache-airflow-providers-microsoft-azure==4.3.0
apache-airflow-providers-mysql==3.2.1
apache-airflow-providers-odbc==3.1.2
apache-airflow-providers-postgres==5.2.2
apache-airflow-providers-presto==4.2.0
apache-airflow-providers-redis==3.0.0
apache-airflow-providers-sendgrid==3.0.0
apache-airflow-providers-sftp==4.1.0
apache-airflow-providers-slack==6.0.0
apache-airflow-providers-sqlite==3.2.1
apache-airflow-providers-ssh==3.2.0

### Deployment

Other 3rd-party Helm chart

### Deployment details

_No response_

### Anything else

_No response_

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `airflow/providers/google/cloud/operators/cloud_base.py`
Content:
```
1 #
2 # Licensed to the Apache Software Foundation (ASF) under one
3 # or more contributor license agreements.  See the NOTICE file
4 # distributed with this work for additional information
5 # regarding copyright ownership.  The ASF licenses this file
6 # to you under the Apache License, Version 2.0 (the
7 # "License"); you may not use this file except in compliance
8 # with the License.  You may obtain a copy of the License at
9 #
10 #   http://www.apache.org/licenses/LICENSE-2.0
11 #
12 # Unless required by applicable law or agreed to in writing,
13 # software distributed under the License is distributed on an
14 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 # KIND, either express or implied.  See the License for the
16 # specific language governing permissions and limitations
17 # under the License.
18 """This module contains a Google API base operator."""
19 from __future__ import annotations
20 
21 from airflow.models import BaseOperator
22 
23 
24 class GoogleCloudBaseOperator(BaseOperator):
25     """
26     Abstract base class that takes care of common specifics of the operators built
27     on top of Google API client libraries.
28     """
29 
30     pass
31
```
--- 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/airflow/providers/google/cloud/operators/cloud_base.py b/airflow/providers/google/cloud/operators/cloud_base.py
--- a/airflow/providers/google/cloud/operators/cloud_base.py
+++ b/airflow/providers/google/cloud/operators/cloud_base.py
@@ -18,6 +18,8 @@
 """This module contains a Google API base operator."""
 from __future__ import annotations
 
+from google.api_core.gapic_v1.method import DEFAULT
+
 from airflow.models import BaseOperator
 
 
@@ -27,4 +29,11 @@
     on top of Google API client libraries.
     """
 
-    pass
+    def __deepcopy__(self, memo):
+        """
+        Updating the memo to fix the non-copyable global constant.
+        This constant can be specified in operator parameters as a retry configuration to indicate a default.
+        See https://github.com/apache/airflow/issues/28751 for details.
+        """
+        memo[id(DEFAULT)] = DEFAULT
+        return super().__deepcopy__(memo)

{"golden_diff": "diff --git a/airflow/providers/google/cloud/operators/cloud_base.py b/airflow/providers/google/cloud/operators/cloud_base.py\n--- a/airflow/providers/google/cloud/operators/cloud_base.py\n+++ b/airflow/providers/google/cloud/operators/cloud_base.py\n@@ -18,6 +18,8 @@\n \"\"\"This module contains a Google API base operator.\"\"\"\n from __future__ import annotations\n \n+from google.api_core.gapic_v1.method import DEFAULT\n+\n from airflow.models import BaseOperator\n \n \n@@ -27,4 +29,11 @@\n     on top of Google API client libraries.\n     \"\"\"\n \n-    pass\n+    def __deepcopy__(self, memo):\n+        \"\"\"\n+        Updating the memo to fix the non-copyable global constant.\n+        This constant can be specified in operator parameters as a retry configuration to indicate a default.\n+        See https://github.com/apache/airflow/issues/28751 for details.\n+        \"\"\"\n+        memo[id(DEFAULT)] = DEFAULT\n+        return super().__deepcopy__(memo)\n", "issue": "KubernetesExecutor leaves failed pods due to deepcopy issue with Google providers\n### Apache Airflow version\r\n\r\nOther Airflow 2 version (please specify below)\r\n\r\n### What happened\r\n\r\nWith Airflow 2.3 and 2.4 there appears to be a bug in the KubernetesExecutor when used in conjunction with the Google airflow providers. This bug does not affect Airflow 2.2 due to the pip version requirements.\r\n\r\nThe bug specifically presents itself when using nearly any Google provider operator. During the pod lifecycle, all is well until the executor in the pod starts to clean up following a successful run. Airflow itself still see's the task marked as a success, but in Kubernetes, while the task is finishing up after reporting status, it actually crashes and puts the pod into a Failed state silently:\r\n```\r\nTraceback (most recent call last):\r\n  File \"/home/airflow/.local/bin/airflow\", line 8, in <module>\r\n    sys.exit(main())\r\n  File \"/home/airflow/.local/lib/python3.9/site-packages/airflow/__main__.py\", line 39, in main\r\n    args.func(args)\r\n  File \"/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/cli_parser.py\", line 52, in command\r\n    return func(*args, **kwargs)\r\n  File \"/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/cli.py\", line 103, in wrapper\r\n    return f(*args, **kwargs)\r\n  File \"/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/commands/task_command.py\", line 382, in task_run\r\n    _run_task_by_selected_method(args, dag, ti)\r\n  File \"/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/commands/task_command.py\", line 189, in _run_task_by_selected_method\r\n    _run_task_by_local_task_job(args, ti)\r\n  File \"/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/commands/task_command.py\", line 247, in _run_task_by_local_task_job\r\n    run_job.run()\r\n  File \"/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/base_job.py\", line 247, in run\r\n    self._execute()\r\n  File \"/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/local_task_job.py\", line 137, in _execute\r\n    self.handle_task_exit(return_code)\r\n  File \"/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/local_task_job.py\", line 168, in handle_task_exit\r\n    self._run_mini_scheduler_on_child_tasks()\r\n  File \"/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/session.py\", line 75, in wrapper\r\n    return func(*args, session=session, **kwargs)\r\n  File \"/home/airflow/.local/lib/python3.9/site-packages/airflow/jobs/local_task_job.py\", line 253, in _run_mini_scheduler_on_child_tasks\r\n    partial_dag = task.dag.partial_subset(\r\n  File \"/home/airflow/.local/lib/python3.9/site-packages/airflow/models/dag.py\", line 2188, in partial_subset\r\n    dag.task_dict = {\r\n  File \"/home/airflow/.local/lib/python3.9/site-packages/airflow/models/dag.py\", line 2189, in <dictcomp>\r\n    t.task_id: _deepcopy_task(t)\r\n  File \"/home/airflow/.local/lib/python3.9/site-packages/airflow/models/dag.py\", line 2186, in _deepcopy_task\r\n    return copy.deepcopy(t, memo)\r\n  File \"/usr/local/lib/python3.9/copy.py\", line 153, in deepcopy\r\n    y = copier(memo)\r\n  File \"/home/airflow/.local/lib/python3.9/site-packages/airflow/models/baseoperator.py\", line 1163, in __deepcopy__\r\n    setattr(result, k, copy.deepcopy(v, memo))\r\n  File \"/usr/local/lib/python3.9/copy.py\", line 172, in deepcopy\r\n    y = _reconstruct(x, memo, *rv)\r\n  File \"/usr/local/lib/python3.9/copy.py\", line 264, in _reconstruct\r\n    y = func(*args)\r\n  File \"/usr/local/lib/python3.9/enum.py\", line 384, in __call__\r\n    return cls.__new__(cls, value)\r\n  File \"/usr/local/lib/python3.9/enum.py\", line 702, in __new__\r\n    raise ve_exc\r\nValueError: <object object at 0x7f570181a3c0> is not a valid _MethodDefault\r\n```\r\n\r\nBased on a quick look, it appears to be related to the default argument that Google is using in its operators which happens to be an Enum, and fails during a deepcopy at the end of the task.\r\n\r\nExample operator that is affected: https://github.com/apache/airflow/blob/403ed7163f3431deb7fc21108e1743385e139907/airflow/providers/google/cloud/hooks/dataproc.py#L753\r\nReference to the Google Python API core which has the Enum causing the problem: https://github.com/googleapis/python-api-core/blob/main/google/api_core/gapic_v1/method.py#L31\r\n\r\n### What you think should happen instead\r\n\r\nKubernetes pods should succeed, be marked as `Completed`, and then be gracefully terminated.\r\n\r\n### How to reproduce\r\n\r\nUse any `apache-airflow-providers-google` >= 7.0.0 which includes `google-api-core` >= 2.2.2. Run a DAG with a task which uses any of the Google operators which have `_MethodDefault` as a default argument.\r\n\r\n### Operating System\r\n\r\nDebian GNU/Linux 11 (bullseye)\r\n\r\n### Versions of Apache Airflow Providers\r\n\r\napache-airflow-providers-amazon==6.0.0\r\napache-airflow-providers-apache-hive==5.0.0\r\napache-airflow-providers-celery==3.0.0\r\napache-airflow-providers-cncf-kubernetes==4.4.0\r\napache-airflow-providers-common-sql==1.3.1\r\napache-airflow-providers-docker==3.2.0\r\napache-airflow-providers-elasticsearch==4.2.1\r\napache-airflow-providers-ftp==3.1.0\r\napache-airflow-providers-google==8.4.0\r\napache-airflow-providers-grpc==3.0.0\r\napache-airflow-providers-hashicorp==3.1.0\r\napache-airflow-providers-http==4.0.0\r\napache-airflow-providers-imap==3.0.0\r\napache-airflow-providers-microsoft-azure==4.3.0\r\napache-airflow-providers-mysql==3.2.1\r\napache-airflow-providers-odbc==3.1.2\r\napache-airflow-providers-postgres==5.2.2\r\napache-airflow-providers-presto==4.2.0\r\napache-airflow-providers-redis==3.0.0\r\napache-airflow-providers-sendgrid==3.0.0\r\napache-airflow-providers-sftp==4.1.0\r\napache-airflow-providers-slack==6.0.0\r\napache-airflow-providers-sqlite==3.2.1\r\napache-airflow-providers-ssh==3.2.0\r\n\r\n### Deployment\r\n\r\nOther 3rd-party Helm chart\r\n\r\n### Deployment details\r\n\r\n_No response_\r\n\r\n### Anything else\r\n\r\n_No response_\r\n\r\n### Are you willing to submit PR?\r\n\r\n- [X] Yes I am willing to submit a PR!\r\n\r\n### Code of Conduct\r\n\r\n- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)\r\n\n", "before_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  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,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\"\"\"This module contains a Google API base operator.\"\"\"\nfrom __future__ import annotations\n\nfrom airflow.models import BaseOperator\n\n\nclass GoogleCloudBaseOperator(BaseOperator):\n    \"\"\"\n    Abstract base class that takes care of common specifics of the operators built\n    on top of Google API client libraries.\n    \"\"\"\n\n    pass\n", "path": "airflow/providers/google/cloud/operators/cloud_base.py"}], "after_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  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,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\"\"\"This module contains a Google API base operator.\"\"\"\nfrom __future__ import annotations\n\nfrom google.api_core.gapic_v1.method import DEFAULT\n\nfrom airflow.models import BaseOperator\n\n\nclass GoogleCloudBaseOperator(BaseOperator):\n    \"\"\"\n    Abstract base class that takes care of common specifics of the operators built\n    on top of Google API client libraries.\n    \"\"\"\n\n    def __deepcopy__(self, memo):\n        \"\"\"\n        Updating the memo to fix the non-copyable global constant.\n        This constant can be specified in operator parameters as a retry configuration to indicate a default.\n        See https://github.com/apache/airflow/issues/28751 for details.\n        \"\"\"\n        memo[id(DEFAULT)] = DEFAULT\n        return super().__deepcopy__(memo)\n", "path": "airflow/providers/google/cloud/operators/cloud_base.py"}]}