Revert "Remove remaining Airflow 2.5 backcompat code from Google Prov…

[potiuk]

…ider (#36366)"

This reverts commit 2c2763f.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[potiuk]

I think (though I have no idea how) this one triggered the (pretty repeatable) failure: https://github.com/apache/airflow/actions/runs/7331844040/job/19966037471 :

=================================== FAILURES ===================================
____________ TestGetGcpCredentialsAndProjectId.test_disable_logging ____________

self = <tests.providers.google.cloud.utils.test_credentials_provider.TestGetGcpCredentialsAndProjectId object at 0x7f9947f85e10>
mock_default = <MagicMock name='from_service_account_file' id='140296134658464'>
mock_info = <MagicMock name='from_service_account_info' id='140296134666336'>
mock_file = <MagicMock name='default' id='140296134706992'>
caplog = <_pytest.logging.LogCaptureFixture object at 0x7f993d450880>

    @mock.patch("google.auth.default", return_value=("CREDENTIALS", "PROJECT_ID"))
    @mock.patch(
        "google.oauth2.service_account.Credentials.from_service_account_info",
    )
    @mock.patch(
        "google.oauth2.service_account.Credentials.from_service_account_file",
    )
    def test_disable_logging(self, mock_default, mock_info, mock_file, caplog):
        """Test disable logging in ``get_credentials_and_project_id``"""
    
        # assert no logs
        with caplog.at_level(level=logging.DEBUG, logger=CRED_PROVIDER_LOGGER_NAME):
            caplog.clear()
            get_credentials_and_project_id(disable_logging=True)
            assert not caplog.record_tuples
    
        # assert no debug logs emitted from get_credentials_and_project_id
        with caplog.at_level(level=logging.DEBUG, logger=CRED_PROVIDER_LOGGER_NAME):
            caplog.clear()
            get_credentials_and_project_id(
                keyfile_dict={"private_key": "PRIVATE_KEY"},
                disable_logging=True,
            )
            assert not caplog.record_tuples
    
        # assert no debug logs emitted from get_credentials_and_project_id
        with caplog.at_level(level=logging.DEBUG, logger=CRED_PROVIDER_LOGGER_NAME):
            caplog.clear()
            get_credentials_and_project_id(
                key_path="KEY.json",
                disable_logging=True,
            )
>           assert not caplog.record_tuples
E           assert not [('asyncio', 40, "Task was destroyed but it is pending!\ntask: <Task pending name='Task-73' coro=<<async_generator_asend without __name__>()> wait_for=<Future pending cb=[Task.task_wakeup()]>>")]
E            +  where [('asyncio', 40, "Task was destroyed but it is pending!\ntask: <Task pending name='Task-73' coro=<<async_generator_asend without __name__>()> wait_for=<Future pending cb=[Task.task_wakeup()]>>")] = <_pytest.logging.LogCaptureFixture object at 0x7f993d450880>.record_tuples

tests/providers/google/cloud/utils/test_credentials_provider.py:392: AssertionError
----------------------------- Captured stderr call -----------------------------
ERROR [asyncio] Task was destroyed but it is pending!
task: <Task pending name='Task-73' coro=<<async_generator_asend without __name__>()> wait_for=<Future pending cb=[Task.task_wakeup()]>>
------------------------------ Captured log call -------------------------------
ERROR    asyncio:base_events.py:1758 Task was destroyed but it is pending!
task: <Task pending name='Task-73' coro=<<async_generator_asend without __name__>()> wait_for=<Future pending cb=[Task.task_wakeup()]>>
----- generated xml file: /files/test_result-providers_google-postgres.xml -----

I don't think this is a direct cause - the error looks like a side-effect of some asyncio test - i just want to see if reverting it will bring the green status again (seems that it started to happen after this one was merged).

[potiuk]

It's a very strange one -- very consistently happening for Postgres 14 + Pyhon 3.10 (and only for that combination (!??!)... If that one will not fail, I plan to merge it and then we can try to start looking for the reason.

[potiuk]

Yep. Looks like reverting this change brings bavck "green" status.

[hussein-awala]

LGTM, I will take a look once this one is merged.

[potiuk]

I believe this error is manifestation of this pytest-asyncIo issue: pytest-dev/pytest-asyncio#200

It looks like the way how pytest-asyncio implements their pytest fixtures is that it (sometimes) creates a new event loop to handle tests and when there are some pending tasks those tests are not cleaned immediately when the fixture event loop is leaving the test scope. So seems that the change in airflow/providers/google/cloud/log/stackdriver_task_handler.py (this is where asyncio code is used) is triggering a condition where stackdriver's test is leaving an asyncio task behind and that task only gets cancelled much later - and accidentally it produces a log entry in the same logger that is then verified in test_disable_logging .

[potiuk]

Added #36442 to test the hypothesis