Fix impersonate beam condition lower than 2.39.0 always evaluating true

[tiranux]

Impersonation of service accounts was added in Apache Beam 2.39.0. In PR #27263 validation was added to fix backwards compatibility, but there is an "or True" condition that causes to always evaluate to true. This is causing environment with Apache Beam 2.39.0 and higher to always raise the exception. Same condition was replicated in PR #31471 when support for deferrable mode was added.

Removing conflicting "or True" from code:

airflow/airflow/providers/apache/beam/hooks/beam.py

Line 276 in b5a4d36

if Version(beam_version) < Version("2.39.0") or True:

airflow/airflow/providers/apache/beam/hooks/beam.py

Line 491 in b5a4d36

if Version(beam_version) < Version("2.39.0") or True:

---

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

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: [email protected]
  Slack: https://s.apache.org/airflow-slack

[potiuk]

@VladaZakharova -> do you know of any reason where or True was there?

[tiranux]

@VladaZakharova -> do you know of any reason where or True was there?

Maybe @mik-laj can let us know if there was something else there that did not make it into the final commit.

[VladaZakharova]

Hi Team!
I was not really suspicious about this condition on Beam version, because for me, even using version 2.46.0 didn't cause any errors at all:

Please, correct me if i am running the operator not in correct way to reproduce the error:

 start_python_pipeline_dataflow_runner = BeamRunPythonPipelineOperator(
        task_id="start_python_pipeline_dataflow_runner",
        runner="DataflowRunner",
        py_file=GCS_PYTHON_SCRIPT,
        pipeline_options={
            "tempLocation": GCS_TMP,
            "stagingLocation": GCS_STAGING,
            "output": GCS_OUTPUT,
        },
        py_options=[],
        py_requirements=["apache-beam[gcp]==2.46.0"],
        py_interpreter="python3",
        py_system_site_packages=False,
        dataflow_config=DataflowConfiguration(
            job_name="{{task.task_id}}", project_id=GCP_PROJECT_ID, location="us-central1"
        ),
        deferrable=True,
    )

[tiranux]

@VladaZakharova your example is not using impersonate_service_account, so it does not enter into that condition:

airflow/airflow/providers/apache/beam/hooks/beam.py

Lines 274 to 279 in b5a4d36

	impersonate_service_account = variables.get("impersonate_service_account")
	if impersonate_service_account:
	if Version(beam_version) < Version("2.39.0") or True:
	raise AirflowException(
	"The impersonateServiceAccount option requires Apache Beam 2.39.0 or newer."
	)

[VladaZakharova]

Sorry, my bad :D
Using this parameter it failed for me as well.
I am not sure about the origin of this additional check in if statement, but i think it is also a good idea to remove it.

[potiuk]

Looks like left-over debugging :)

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.