Fix Google Cloud Data Fusion hook to handle pipeline start errors properly

[alamashir]

Description

Fixes #50387 - Bug in Data Fusion hook where start_pipeline crashes with KeyError: 'runId' when pipelines fail to start.

Problem

The hook used the multi-program start endpoint which returns HTTP 200 even on failures. The code accessed response_json[0]["runId"] without validation, causing a KeyError when the field was missing.

Solution

• Switch to single-program start endpoint: POST .../apps/{app}/{program-type}s/{program-id}/start
• Validate runId exists in response before accessing
• Return clear error messages when pipelines fail to start

Changes

Updated start_pipeline method to use correct CDAP endpoint and added response validation.

Breaking Changes

None - method signature and return type unchanged. Only the internal API endpoint changed.

Testing

pytest providers/google/tests/unit/google/cloud/hooks/test_datafusion.py -v

[VladaZakharova]

hi there! thank you for the fix. Can please also provide the screenshot of the green system tests running in Airflow UI?

[alamashir]

@VladaZakharova Thank you for the feedback! I've verified the fix works correctly with all 45 unit tests passing, including a new test specifically for the error case (test_start_pipeline_should_fail_if_no_run_id).

Running the full system tests requires setting up and provisioning a Data Fusion instance. Given that the unit tests comprehensively cover the fix (switching from multi-program to single-program endpoint + runId validation), I believe they provide sufficient verification. However, if you'd like me to set up and run the full system tests, I'm happy to do so.

[shahar1]

owever, if you'd like me to set up and run the full system tests, I'm happy to do so.

@VladaZakharova Thank you for the feedback! I've verified the fix works correctly with all 45 unit tests passing, including a new test specifically for the error case (test_start_pipeline_should_fail_if_no_run_id).

Running the full system tests requires setting up and provisioning a Data Fusion instance. Given that the unit tests comprehensively cover the fix (switching from multi-program to single-program endpoint + runId validation), I believe they provide sufficient verification. However, if you'd like me to set up and run the full system tests, I'm happy to do so.

If you're able to run the system tests on your side and provide some screenshots, we'll be happy if you could do so.
If not, please let us know - I believe that Vlada's team could figure out a solution for that.

[alamashir]

@shahar1 @VladaZakharova if you can, that will be great.
I spent some time setting up CDF session but had some issues and that will need quite some time for me to setup. I can spend more time if needed end of next week. So if i dont hear back from you all by then, i can do it.