Issue 1:
Apache Airflow Provider(s)

google

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==6.2.0

apache-airflow-providers-celery==3.1.0

apache-airflow-providers-cncf-kubernetes==5.0.0

apache-airflow-providers-common-sql==1.3.1

apache-airflow-providers-databricks==3.3.0

apache-airflow-providers-docker==3.3.0

apache-airflow-providers-elasticsearch==4.3.1

apache-airflow-providers-ftp==3.2.0

apache-airflow-providers-google==8.3.0

apache-airflow-providers-grpc==3.1.0

apache-airflow-providers-hashicorp==3.2.0

apache-airflow-providers-http==4.1.0

apache-airflow-providers-imap==3.1.0

apache-airflow-providers-microsoft-azure==5.0.0

apache-airflow-providers-odbc==3.2.1

apache-airflow-providers-postgres==5.3.1

apache-airflow-providers-redis==3.1.0

apache-airflow-providers-sendgrid==3.1.0

apache-airflow-providers-sftp==4.2.0

apache-airflow-providers-slack==7.1.0

apache-airflow-providers-sqlite==3.3.1

apache-airflow-providers-ssh==3.3.0

Apache Airflow version

2.5.0

Operating System

Debian GNU/Linux 11 (bullseye)

Deployment

Official Apache Airflow Helm Chart

Deployment details

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.

What happened

On February 1 2023, Google Ads sunset Google Ads API v10 and other versions are going to be discontinued soon. Currently, the lowest Google Ads API version that offers a reasonable amount of availability time is v12, and the lowest version of the google-ads package that offers support for API v12 is 19.0.0 (the package is currently at version 20.0.0).

Using version 8.3.0 of apache-airflow-providers-google package, trying to bump google-adsup to 19.0.0 creates a dependency hell with other Google packages, and upgrading to 20.0.0 causes the following error:

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 "<frozen importlib._bootstrap>", line 1030, in _gcd_import
airflow-dags-airflow-webserver-1 | File "<frozen importlib._bootstrap>", line 1007, in _find_and_load
airflow-dags-airflow-webserver-1 | File "<frozen importlib._bootstrap>", line 986, in _find_and_load_unlocked
airflow-dags-airflow-webserver-1 | File "<frozen importlib._bootstrap>", line 680, in _load_unlocked
airflow-dags-airflow-webserver-1 | File "<frozen importlib._bootstrap_external>", line 850, in exec_module
airflow-dags-airflow-webserver-1 | File "<frozen importlib._bootstrap>", 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 <module>
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 <module>
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 <module>
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 <module>
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 <module>
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 <module>
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 <module>
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 <module>
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 >= 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).

Following versions of the provider package can't resolve dependencies when installing requirements, because of PR #25886 (found here, 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.

We cannot downgrade protobuf because google-ads requires >=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.

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.

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).

What you think should happen instead

No response

How to reproduce

Install apache-airflow-providers-google version 8.3.0

Install google-ads version 20.0.0

Try to load any DAG that uses any Google operator or imports anything from Google's provider package

Anything else

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct

Issue 2:
This test is flaky and fails sometimes

____________ TestLocalTaskJob.test_task_sigkill_works_with_retries _____________

self = <tests.jobs.test_local_task_job.TestLocalTaskJob object at 0x7f3a8ec6c6a0>
dag_maker = <tests.conftest.dag_maker.<locals>.DagFactory object at 0x7f3aa1490080>

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()
> assert ti.state == State.UP_FOR_RETRY
E AssertionError: assert None == <TaskInstanceState.UP_FOR_RETRY: 'up_for_retry'>
E + where None = <TaskInstance: test_mark_failure_2.test_on_failure 2016-01-01 00:00:00+00:00 [None]>.state
E + and <TaskInstanceState.UP_FOR_RETRY: 'up_for_retry'> = State.UP_FOR_RETRY

tests/jobs/test_local_task_job.py:778: AssertionError
----------------------------- Captured stdout call -----------------------------
Running <TaskInstance: test_mark_failure_2.test_on_failure 2016-01-01T00:00:00+00:00 [None]> 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'")