BeamRunPythonPipelineOperator does not set gcp_conn_id correctly

[nan0o]

Apache Airflow Provider(s)

apache-beam

Versions of Apache Airflow Providers

apache-airflow-providers-apache-beam==4.0.0
apache-airflow-providers-cncf-kubernetes==4.3.0
apache-airflow-providers-dbt-cloud==2.1.0
apache-airflow-providers-ftp==3.1.0
apache-airflow-providers-google==8.1.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-mysql==3.0.0
apache-airflow-providers-postgres==5.0.0
apache-airflow-providers-sendgrid==3.0.0
apache-airflow-providers-sqlite==3.0.0
apache-airflow-providers-ssh==3.1.0
apache-beam==2.41.0

Apache Airflow version

2.3.3

Operating System

Not sure what OS GKE Pods use

Deployment

Composer

Deployment details

No response

What happened

When using BeamRunPythonPipelineOperator, composer shows in the task logs that it's trying to use "google_cloud_default" as a connection id even though I specify in the operator that it should be "dataflow_conn"

What you think should happen instead

Use "dataflow_conn" as the connection id

How to reproduce

BeamRunPythonPipelineOperator(
    task_id='xxxx',
    runner="DataflowRunner",
    # This doesn't work even though the docs
    # indicate you should pass this to the constructor
    #gcp_conn_id='dataflow_conn',
    py_file=f'/home/airflow/gcs/data/{project_name}/main.py',
    pipeline_options={
        'project_id': project_id,
        'input_path': input_path,
        'output_path': output_path,
        'schema_path': schema_path,
        'service_account': service_account,     
        'no_use_public_ips': True,
        'subnetwork': subnetwork,      
        'staging_location': staging_location,
        'temp_location': temp_location,
        "setup_file": f'/home/airflow/gcs/data/{project_name}/setup.py',
        "machine_type": "n1-standard-4",
        "num_workers": 5,
        "max_num_workers": 10,
    },
    py_options=[],
    py_interpreter='python3',
    py_system_site_packages=False,
    dataflow_config=DataflowConfiguration(
        job_name='{{task.task_id}}',
        location=gce_region,
        wait_until_finished=False,
        # This works for some reason!
        gcp_conn_id="dataflow_conn"
    ),
)

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

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template!

[potiuk]

I think you made some typo. Not reproducible. You need to be more specific including showing exact logs (and ideally try with latest provider first - no idea which provider version is by default used in composer).

[nan0o]

Tomorrow I will post the logs, I do not think it was a typo. As for the providers, I am not sure how to check the version in composer.

[potiuk]

You can check it in the UI->Admin

[nan0o]

I will post images so there's no room for error while transcribing messages. This is an image of my DAG code:

For BeamRunPythonPipelineOperator, I get this message that it's using 'google_cloud_default':

Final traceback:

[2022-11-11, 14:44:19 UTC] {beam.py:127} WARNING - apitools.base.py.exceptions.HttpForbiddenError: HttpError accessing <https://dataflow.googleapis.com/v1b3/projects/<project>/locations/us-east1/jobs?alt=json>: response: <{'vary': 'Origin, X-Origin, Referer', 'content-type': 'application/json; charset=UTF-8', 'date': 'Fri, 11 Nov 2022 14:44:19 GMT', 'server': 'ESF', 'cache-control': 'private', 'x-xss-protection': '0', 'x-frame-options': 'SAMEORIGIN', 'x-content-type-options': 'nosniff', 'transfer-encoding': 'chunked', 'status': '403', 'content-length': '314', '-content-encoding': 'gzip'}>, content <{
[2022-11-11, 14:44:19 UTC] {beam.py:127} WARNING -   "error": {
[2022-11-11, 14:44:19 UTC] {beam.py:127} WARNING -     "code": 403,
[2022-11-11, 14:44:19 UTC] {beam.py:127} WARNING -     "message": "(cb5a6c54e7d87a4d): Could not create workflow; user does not have write access to project: <project> Causes: (cb5a6c54e7d87246): Permission 'dataflow.jobs.create' denied on project: '<project>'",
[2022-11-11, 14:44:19 UTC] {beam.py:127} WARNING -     "status": "PERMISSION_DENIED"
[2022-11-11, 14:44:19 UTC] {beam.py:127} WARNING -   }
[2022-11-11, 14:44:19 UTC] {beam.py:127} WARNING - }

And here's an image of my dataflow_conn connection id.

I would now do the same showing that it works putting it in DataflowConfiguration but I raised this issue since it didn't work from the constructor as the docs say.

[nan0o]

Also, I found the official documentation for composer versions and packages that come shipped with it:
composer-2.0.28-airflow-2.3.3

apache-airflow-providers-apache-beam==4.0.0
apache-airflow-providers-cncf-kubernetes==4.3.0
apache-airflow-providers-dbt-cloud==2.1.0
apache-airflow-providers-ftp==3.1.0
apache-airflow-providers-google==8.1.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-mysql==3.0.0
apache-airflow-providers-postgres==5.0.0
apache-airflow-providers-sendgrid==3.0.0
apache-airflow-providers-sqlite==3.0.0
apache-airflow-providers-ssh==3.1.0
apache-beam==2.41.0

[potiuk]

Converted to discussion.