Fix Beam Dataflow deferral without launcher job id

[fat-catTW]

Why

Beam Dataflow operators relied on Beam launcher logs to discover the Dataflow job id before deferring. When those logs omit the id, the deferrable path could pass None into the Dataflow trigger and fail instead of using the existing job-name based wait path.

relates: #68279

Solution

Try to resolve the Dataflow job id by job name while the Beam launcher is still running. If the id is found, deferrable execution proceeds with a valid trigger job id. If the id still cannot be resolved, fall back to the synchronous wait_for_done path instead of deferring with None.

Was generative AI tooling used to co-author this PR?

• [X] Yes (please specify the tool below)

Generated-by: [Codex] following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[MaksYermak]

@fat-catTW how did you check these changes? Because in the current time apache-beam-provider in the suspend state and all tests(unit, system) are disabled in the CI.

In the current time I am working on unsuspend beam provider in this PR. Also the fix for the issue, which you mentioned, was prepared by me in this PR and wait when beam provider will be unsuspend.

[fat-catTW]

Thanks for pointing this out.
I confirmed that the Apache Beam provider tests are currently excluded from CI while the provider is suspended, and the related Google Dataflow test module is skipped as well.

I also reviewed #66952 and the existing fix in VladaZakharova#325. Although this PR takes a somewhat different approach, both changes address the same underlying issue.

Since the existing work started earlier and is already coordinated with the Beam provider unsuspension, I’m converting this PR to draft and will leave the issue to that work.

Please feel free to reference or reuse anything useful from this implementation, particularly the periodic job ID lookup while the launcher is still producing output, the synchronous fallback when no job ID can be resolved, and the additional Beam operator tests.

Thanks for the clarification.

[fat-catTW]

Maybe, one way to validate the fix locally before #66952 is merged would be to create a temporary worktree based on #66952 and apply the fix on top of it?
This should allow the suspended Beam tests to run instead of being skipped.