<h3 dir="auto">Apache Airflow Provider(s)</h3>
<p dir="auto">google</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow-providers-amazon==6.2.0<br>
apache-airflow-providers-celery==3.1.0<br>
apache-airflow-providers-cncf-kubernetes==5.0.0<br>
apache-airflow-providers-common-sql==1.3.1<br>
apache-airflow-providers-databricks==3.3.0<br>
apache-airflow-providers-docker==3.3.0<br>
apache-airflow-providers-elasticsearch==4.3.1<br>
apache-airflow-providers-ftp==3.2.0<br>
apache-airflow-providers-google==8.3.0<br>
apache-airflow-providers-grpc==3.1.0<br>
apache-airflow-providers-hashicorp==3.2.0<br>
apache-airflow-providers-http==4.1.0<br>
apache-airflow-providers-imap==3.1.0<br>
apache-airflow-providers-microsoft-azure==5.0.0<br>
apache-airflow-providers-odbc==3.2.1<br>
apache-airflow-providers-postgres==5.3.1<br>
apache-airflow-providers-redis==3.1.0<br>
apache-airflow-providers-sendgrid==3.1.0<br>
apache-airflow-providers-sftp==4.2.0<br>
apache-airflow-providers-slack==7.1.0<br>
apache-airflow-providers-sqlite==3.3.1<br>
apache-airflow-providers-ssh==3.3.0</p>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.5.0</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Debian GNU/Linux 11 (bullseye)</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Official Apache Airflow Helm Chart</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">We deploy the official Airflow helm chart on Kubernetes on GKE using ArgoCD. We use a custom image derived from the provided official Docker image where the only additional step is the installation of requirements.txt.</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">On February 1 2023, Google Ads sunset Google Ads API v10 and other versions are going to be <a href="https://developers.google.com/google-ads/api/docs/sunset-dates" rel="nofollow">discontinued soon</a>. Currently, the lowest Google Ads API version that offers a reasonable amount of availability time is v12, and the lowest version of the <code class="notranslate">google-ads</code> package that offers support for API v12 is 19.0.0 (the package is currently at version 20.0.0).</p>
<p dir="auto">Using version 8.3.0 of  <code class="notranslate">apache-airflow-providers-google</code> package, trying to bump <code class="notranslate">google-ads</code>up  to 19.0.0 creates a dependency hell with other Google packages, and upgrading to 20.0.0 causes the following error:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="airflow-dags-airflow-webserver-1  | Traceback (most recent call last):
airflow-dags-airflow-webserver-1  |   File &quot;/home/airflow/.local/lib/python3.9/site-packages/airflow/providers_manager.py&quot;, line 275, in _sanity_check
airflow-dags-airflow-webserver-1  |     imported_class = import_string(class_name)
airflow-dags-airflow-webserver-1  |   File &quot;/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/module_loading.py&quot;, line 33, in import_string
airflow-dags-airflow-webserver-1  |     module = import_module(module_path)
airflow-dags-airflow-webserver-1  |   File &quot;/usr/local/lib/python3.9/importlib/__init__.py&quot;, line 127, in import_module
airflow-dags-airflow-webserver-1  |     return _bootstrap._gcd_import(name[level:], package, level)
airflow-dags-airflow-webserver-1  |   File &quot;&lt;frozen importlib._bootstrap&gt;&quot;, line 1030, in _gcd_import
airflow-dags-airflow-webserver-1  |   File &quot;&lt;frozen importlib._bootstrap&gt;&quot;, line 1007, in _find_and_load
airflow-dags-airflow-webserver-1  |   File &quot;&lt;frozen importlib._bootstrap&gt;&quot;, line 986, in _find_and_load_unlocked
airflow-dags-airflow-webserver-1  |   File &quot;&lt;frozen importlib._bootstrap&gt;&quot;, line 680, in _load_unlocked
airflow-dags-airflow-webserver-1  |   File &quot;&lt;frozen importlib._bootstrap_external&gt;&quot;, line 850, in exec_module
airflow-dags-airflow-webserver-1  |   File &quot;&lt;frozen importlib._bootstrap&gt;&quot;, line 228, in _call_with_frames_removed
airflow-dags-airflow-webserver-1  |   File &quot;/home/airflow/.local/lib/python3.9/site-packages/airflow/providers/google/cloud/hooks/compute_ssh.py&quot;, line 26, in &lt;module&gt;
airflow-dags-airflow-webserver-1  |     from airflow.providers.google.cloud.hooks.compute import ComputeEngineHook
airflow-dags-airflow-webserver-1  |   File &quot;/home/airflow/.local/lib/python3.9/site-packages/airflow/providers/google/cloud/hooks/compute.py&quot;, line 26, in &lt;module&gt;
airflow-dags-airflow-webserver-1  |     from airflow.providers.google.common.hooks.base_google import PROVIDE_PROJECT_ID, GoogleBaseHook
airflow-dags-airflow-webserver-1  |   File &quot;/home/airflow/.local/lib/python3.9/site-packages/airflow/providers/google/common/hooks/base_google.py&quot;, line 49, in &lt;module&gt;
airflow-dags-airflow-webserver-1  |     from airflow.providers.google.cloud.utils.credentials_provider import (
airflow-dags-airflow-webserver-1  |   File &quot;/home/airflow/.local/lib/python3.9/site-packages/airflow/providers/google/cloud/utils/credentials_provider.py&quot;, line 36, in &lt;module&gt;
airflow-dags-airflow-webserver-1  |     from airflow.providers.google.cloud._internal_client.secret_manager_client import _SecretManagerClient
airflow-dags-airflow-webserver-1  |   File &quot;/home/airflow/.local/lib/python3.9/site-packages/airflow/providers/google/cloud/_internal_client/secret_manager_client.py&quot;, line 23, in &lt;module&gt;
airflow-dags-airflow-webserver-1  |     from google.cloud.secretmanager_v1 import SecretManagerServiceClient
airflow-dags-airflow-webserver-1  |   File &quot;/home/airflow/.local/lib/python3.9/site-packages/google/cloud/secretmanager_v1/__init__.py&quot;, line 22, in &lt;module&gt;
airflow-dags-airflow-webserver-1  |     from google.cloud.secretmanager_v1 import types
airflow-dags-airflow-webserver-1  |   File &quot;/home/airflow/.local/lib/python3.9/site-packages/google/cloud/secretmanager_v1/types.py&quot;, line 23, in &lt;module&gt;
airflow-dags-airflow-webserver-1  |     from google.cloud.secretmanager_v1.proto import resources_pb2
airflow-dags-airflow-webserver-1  |   File &quot;/home/airflow/.local/lib/python3.9/site-packages/google/cloud/secretmanager_v1/proto/resources_pb2.py&quot;, line 57, in &lt;module&gt;
airflow-dags-airflow-webserver-1  |     _descriptor.EnumValueDescriptor(
airflow-dags-airflow-webserver-1  |   File &quot;/home/airflow/.local/lib/python3.9/site-packages/google/protobuf/descriptor.py&quot;, line 796, in __new__
airflow-dags-airflow-webserver-1  |     _message.Message._CheckCalledFromGeneratedFile()
airflow-dags-airflow-webserver-1  | TypeError: Descriptors cannot not be created directly.
airflow-dags-airflow-webserver-1  | If this call came from a _pb2.py file, your generated code is out of date and must be regenerated with protoc &gt;= 3.19.0.
airflow-dags-airflow-webserver-1  | If you cannot immediately regenerate your protos, some other possible workarounds are:
airflow-dags-airflow-webserver-1  |  1. Downgrade the protobuf package to 3.20.x or lower.
airflow-dags-airflow-webserver-1  |  2. Set PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python (but this will use pure-Python parsing and will be much slower)."><pre class="notranslate"><code class="notranslate">airflow-dags-airflow-webserver-1  | Traceback (most recent call last):
airflow-dags-airflow-webserver-1  |   File "/home/airflow/.local/lib/python3.9/site-packages/airflow/providers_manager.py", line 275, in _sanity_check
airflow-dags-airflow-webserver-1  |     imported_class = import_string(class_name)
airflow-dags-airflow-webserver-1  |   File "/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/module_loading.py", line 33, in import_string
airflow-dags-airflow-webserver-1  |     module = import_module(module_path)
airflow-dags-airflow-webserver-1  |   File "/usr/local/lib/python3.9/importlib/__init__.py", line 127, in import_module
airflow-dags-airflow-webserver-1  |     return _bootstrap._gcd_import(name[level:], package, level)
airflow-dags-airflow-webserver-1  |   File "&lt;frozen importlib._bootstrap&gt;", line 1030, in _gcd_import
airflow-dags-airflow-webserver-1  |   File "&lt;frozen importlib._bootstrap&gt;", line 1007, in _find_and_load
airflow-dags-airflow-webserver-1  |   File "&lt;frozen importlib._bootstrap&gt;", line 986, in _find_and_load_unlocked
airflow-dags-airflow-webserver-1  |   File "&lt;frozen importlib._bootstrap&gt;", line 680, in _load_unlocked
airflow-dags-airflow-webserver-1  |   File "&lt;frozen importlib._bootstrap_external&gt;", line 850, in exec_module
airflow-dags-airflow-webserver-1  |   File "&lt;frozen importlib._bootstrap&gt;", line 228, in _call_with_frames_removed
airflow-dags-airflow-webserver-1  |   File "/home/airflow/.local/lib/python3.9/site-packages/airflow/providers/google/cloud/hooks/compute_ssh.py", line 26, in &lt;module&gt;
airflow-dags-airflow-webserver-1  |     from airflow.providers.google.cloud.hooks.compute import ComputeEngineHook
airflow-dags-airflow-webserver-1  |   File "/home/airflow/.local/lib/python3.9/site-packages/airflow/providers/google/cloud/hooks/compute.py", line 26, in &lt;module&gt;
airflow-dags-airflow-webserver-1  |     from airflow.providers.google.common.hooks.base_google import PROVIDE_PROJECT_ID, GoogleBaseHook
airflow-dags-airflow-webserver-1  |   File "/home/airflow/.local/lib/python3.9/site-packages/airflow/providers/google/common/hooks/base_google.py", line 49, in &lt;module&gt;
airflow-dags-airflow-webserver-1  |     from airflow.providers.google.cloud.utils.credentials_provider import (
airflow-dags-airflow-webserver-1  |   File "/home/airflow/.local/lib/python3.9/site-packages/airflow/providers/google/cloud/utils/credentials_provider.py", line 36, in &lt;module&gt;
airflow-dags-airflow-webserver-1  |     from airflow.providers.google.cloud._internal_client.secret_manager_client import _SecretManagerClient
airflow-dags-airflow-webserver-1  |   File "/home/airflow/.local/lib/python3.9/site-packages/airflow/providers/google/cloud/_internal_client/secret_manager_client.py", line 23, in &lt;module&gt;
airflow-dags-airflow-webserver-1  |     from google.cloud.secretmanager_v1 import SecretManagerServiceClient
airflow-dags-airflow-webserver-1  |   File "/home/airflow/.local/lib/python3.9/site-packages/google/cloud/secretmanager_v1/__init__.py", line 22, in &lt;module&gt;
airflow-dags-airflow-webserver-1  |     from google.cloud.secretmanager_v1 import types
airflow-dags-airflow-webserver-1  |   File "/home/airflow/.local/lib/python3.9/site-packages/google/cloud/secretmanager_v1/types.py", line 23, in &lt;module&gt;
airflow-dags-airflow-webserver-1  |     from google.cloud.secretmanager_v1.proto import resources_pb2
airflow-dags-airflow-webserver-1  |   File "/home/airflow/.local/lib/python3.9/site-packages/google/cloud/secretmanager_v1/proto/resources_pb2.py", line 57, in &lt;module&gt;
airflow-dags-airflow-webserver-1  |     _descriptor.EnumValueDescriptor(
airflow-dags-airflow-webserver-1  |   File "/home/airflow/.local/lib/python3.9/site-packages/google/protobuf/descriptor.py", line 796, in __new__
airflow-dags-airflow-webserver-1  |     _message.Message._CheckCalledFromGeneratedFile()
airflow-dags-airflow-webserver-1  | TypeError: Descriptors cannot not be created directly.
airflow-dags-airflow-webserver-1  | If this call came from a _pb2.py file, your generated code is out of date and must be regenerated with protoc &gt;= 3.19.0.
airflow-dags-airflow-webserver-1  | If you cannot immediately regenerate your protos, some other possible workarounds are:
airflow-dags-airflow-webserver-1  |  1. Downgrade the protobuf package to 3.20.x or lower.
airflow-dags-airflow-webserver-1  |  2. Set PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python (but this will use pure-Python parsing and will be much slower).
</code></pre></div>
<p dir="auto">Following versions of the provider package can't resolve dependencies when installing requirements, because of PR <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="1347041866" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/25886" data-hovercard-type="pull_request" data-hovercard-url="/apache/airflow/pull/25886/hovercard" href="https://github.com/apache/airflow/pull/25886">#25886</a> (found <a href="https://github.com/apache/airflow/pull/25886" data-hovercard-type="pull_request" data-hovercard-url="/apache/airflow/pull/25886/hovercard">here</a>, I can't link it for some reason). Limiting protobuf's version was an option back then, but cannot be done now since the only functioning Google Ads version does not support this change.</p>
<p dir="auto">We cannot downgrade protobuf because <code class="notranslate">google-ads</code> requires &gt;=4.21.5, which leaves us either with option 2 (which CAN be done, although I'm unsure of the real impact on performance) or with the need to recompile the code using later versions of protoc.</p>
<p dir="auto">If option 2 is the only viable way, we believe at least the documentation should be updated, since it's mandatory that the environment variable is set (in our case, that would mean messing with our Dockerfile) for DAGs to work. However, this would complicate things a lot.</p>
<p dir="auto">C extensions are not something I completely understand, but from what I could gather, I believe that the code the error logs refer should be recompiled with a later version of the protoc compiler. This is just a cascading consequence of Google discontinuing Ads APIs, and the providers need to keep up (and they seem to be doing it pretty fast).</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">How to reproduce</h3>
<ol dir="auto">
<li>Install <code class="notranslate">apache-airflow-providers-google</code> version 8.3.0</li>
<li>Install <code class="notranslate">google-ads</code> version 20.0.0</li>
<li>Try to load any DAG that uses any Google operator or imports anything from Google's provider package</li>
</ol>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>
<p dir="auto">This test is flaky and fails sometimes</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="  ____________ TestLocalTaskJob.test_task_sigkill_works_with_retries _____________
  
  self = &lt;tests.jobs.test_local_task_job.TestLocalTaskJob object at 0x7f3a8ec6c6a0&gt;
  dag_maker = &lt;tests.conftest.dag_maker.&lt;locals&gt;.DagFactory object at 0x7f3aa1490080&gt;
  
      def test_task_sigkill_works_with_retries(self, dag_maker):
          &quot;&quot;&quot;
          Test that ensures that tasks are retried when they receive sigkill
          &quot;&quot;&quot;
          # use shared memory value so we can properly track value change even if
          # it's been updated across processes.
          retry_callback_called = Value('i', 0)
          task_terminated_externally = Value('i', 1)
          shared_mem_lock = Lock()
      
          def retry_callback(context):
              with shared_mem_lock:
                  retry_callback_called.value += 1
              assert context['dag_run'].dag_id == 'test_mark_failure_2'
      
          def task_function(ti):
              os.kill(os.getpid(), signal.SIGKILL)
              # This should not happen -- the state change should be noticed and the task should get killed
              with shared_mem_lock:
                  task_terminated_externally.value = 0
      
          with dag_maker(
              dag_id='test_mark_failure_2', start_date=DEFAULT_DATE, default_args={'owner': 'owner1'}
          ):
              task = PythonOperator(
                  task_id='test_on_failure',
                  python_callable=task_function,
                  retries=1,
                  retry_delay=timedelta(seconds=2),
                  on_retry_callback=retry_callback,
              )
          ti = TaskInstance(task=task, execution_date=DEFAULT_DATE)
          ti.refresh_from_db()
          job1 = LocalTaskJob(task_instance=ti, ignore_ti_state=True, executor=SequentialExecutor())
          job1.task_runner = StandardTaskRunner(job1)
          job1.task_runner.start()
          settings.engine.dispose()
          process = multiprocessing.Process(target=job1.run)
          process.start()
          time.sleep(0.4)
          process.join(timeout=10)
          ti.refresh_from_db()
  &gt;       assert ti.state == State.UP_FOR_RETRY
  E       AssertionError: assert None == &lt;TaskInstanceState.UP_FOR_RETRY: 'up_for_retry'&gt;
  E        +  where None = &lt;TaskInstance: test_mark_failure_2.test_on_failure 2016-01-01 00:00:00+00:00 [None]&gt;.state
  E        +  and   &lt;TaskInstanceState.UP_FOR_RETRY: 'up_for_retry'&gt; = State.UP_FOR_RETRY
  
  tests/jobs/test_local_task_job.py:778: AssertionError
  ----------------------------- Captured stdout call -----------------------------
  Running &lt;TaskInstance: test_mark_failure_2.test_on_failure 2016-01-01T00:00:00+00:00 [None]&gt; on host 9a517fc8fb98
  ----------------------------- Captured stderr call -----------------------------
  Process Process-116:
  Traceback (most recent call last):
    File &quot;/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/base.py&quot;, line 1277, in _execute_context
      cursor, statement, parameters, context
    File &quot;/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/default.py&quot;, line 608, in do_execute
      cursor.execute(statement, parameters)
    File &quot;/usr/local/lib/python3.6/site-packages/MySQLdb/cursors.py&quot;, line 206, in execute
      res = self._query(query)
    File &quot;/usr/local/lib/python3.6/site-packages/MySQLdb/cursors.py&quot;, line 319, in _query
      db.query(q)
    File &quot;/usr/local/lib/python3.6/site-packages/MySQLdb/connections.py&quot;, line 259, in query
      _mysql.connection.query(self, query)
  MySQLdb._exceptions.IntegrityError: (1062, &quot;Duplicate entry 'test_on_failure-test_mark_failure_2-2016-01-01 00:00:00.000000' for key 'PRIMARY'&quot;)"><pre class="notranslate"><code class="notranslate">  ____________ TestLocalTaskJob.test_task_sigkill_works_with_retries _____________
  
  self = &lt;tests.jobs.test_local_task_job.TestLocalTaskJob object at 0x7f3a8ec6c6a0&gt;
  dag_maker = &lt;tests.conftest.dag_maker.&lt;locals&gt;.DagFactory object at 0x7f3aa1490080&gt;
  
      def test_task_sigkill_works_with_retries(self, dag_maker):
          """
          Test that ensures that tasks are retried when they receive sigkill
          """
          # use shared memory value so we can properly track value change even if
          # it's been updated across processes.
          retry_callback_called = Value('i', 0)
          task_terminated_externally = Value('i', 1)
          shared_mem_lock = Lock()
      
          def retry_callback(context):
              with shared_mem_lock:
                  retry_callback_called.value += 1
              assert context['dag_run'].dag_id == 'test_mark_failure_2'
      
          def task_function(ti):
              os.kill(os.getpid(), signal.SIGKILL)
              # This should not happen -- the state change should be noticed and the task should get killed
              with shared_mem_lock:
                  task_terminated_externally.value = 0
      
          with dag_maker(
              dag_id='test_mark_failure_2', start_date=DEFAULT_DATE, default_args={'owner': 'owner1'}
          ):
              task = PythonOperator(
                  task_id='test_on_failure',
                  python_callable=task_function,
                  retries=1,
                  retry_delay=timedelta(seconds=2),
                  on_retry_callback=retry_callback,
              )
          ti = TaskInstance(task=task, execution_date=DEFAULT_DATE)
          ti.refresh_from_db()
          job1 = LocalTaskJob(task_instance=ti, ignore_ti_state=True, executor=SequentialExecutor())
          job1.task_runner = StandardTaskRunner(job1)
          job1.task_runner.start()
          settings.engine.dispose()
          process = multiprocessing.Process(target=job1.run)
          process.start()
          time.sleep(0.4)
          process.join(timeout=10)
          ti.refresh_from_db()
  &gt;       assert ti.state == State.UP_FOR_RETRY
  E       AssertionError: assert None == &lt;TaskInstanceState.UP_FOR_RETRY: 'up_for_retry'&gt;
  E        +  where None = &lt;TaskInstance: test_mark_failure_2.test_on_failure 2016-01-01 00:00:00+00:00 [None]&gt;.state
  E        +  and   &lt;TaskInstanceState.UP_FOR_RETRY: 'up_for_retry'&gt; = State.UP_FOR_RETRY
  
  tests/jobs/test_local_task_job.py:778: AssertionError
  ----------------------------- Captured stdout call -----------------------------
  Running &lt;TaskInstance: test_mark_failure_2.test_on_failure 2016-01-01T00:00:00+00:00 [None]&gt; on host 9a517fc8fb98
  ----------------------------- Captured stderr call -----------------------------
  Process Process-116:
  Traceback (most recent call last):
    File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1277, in _execute_context
      cursor, statement, parameters, context
    File "/usr/local/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 608, in do_execute
      cursor.execute(statement, parameters)
    File "/usr/local/lib/python3.6/site-packages/MySQLdb/cursors.py", line 206, in execute
      res = self._query(query)
    File "/usr/local/lib/python3.6/site-packages/MySQLdb/cursors.py", line 319, in _query
      db.query(q)
    File "/usr/local/lib/python3.6/site-packages/MySQLdb/connections.py", line 259, in query
      _mysql.connection.query(self, query)
  MySQLdb._exceptions.IntegrityError: (1062, "Duplicate entry 'test_on_failure-test_mark_failure_2-2016-01-01 00:00:00.000000' for key 'PRIMARY'")
</code></pre></div>