Skip to content

DataprocCreateClusterOperator: Fix non-deferrable reconciliation and handle deletion during creation - #61951

Merged
potiuk merged 1 commit into
apache:mainfrom
SameerMesiah97:61947-DataprocCreateClusterOperator-Recon-Fix
Mar 10, 2026
Merged

potiuk merged 1 commit into
apache:mainfrom
SameerMesiah97:61947-DataprocCreateClusterOperator-Recon-Fix

Conversation

@SameerMesiah97

Copy link
Copy Markdown
Contributor

Description

This change refactors the DataprocCreateClusterOperator.execute method to ensure cluster state reconciliation is consistently applied in non-deferrable mode.

After submitting the cluster creation request and waiting for the long-running operation (LRO) to complete, the operator now explicitly fetches the current cluster state and passes it through a dedicated _reconcile_cluster_state method before returning success.

The reconciliation logic, previously embedded inline in execute, has been consolidated into _reconcile_cluster_state. This method handles clusters in CREATING, DELETING, and STOPPED states by waiting, recreating, or restarting as appropriate.

Rationale

The operator docstring specifies that when use_if_exists=True, the operator should:

  • Wait if the cluster is in CREATING
  • Wait for deletion and then create a new cluster if in DELETING
  • Handle ERROR state appropriately

Although state-handling logic existed, the non-deferrable execution path previously returned immediately after the create LRO completed, preventing the existing reconciliation logic from being triggered in certain scenarios (e.g. cluster transitioning to DELETING during creation).

This change ensures the pre-existing reconciliation behavior is executed consistently, aligning runtime behavior with the documented contract.

Notes

  • Added explicit NotFound handling after the Long-Running Operation (LRO) completes to surface a clear AirflowException if the cluster was deleted before its state could be reconciled.
  • Additional logging has been added and some existing log messages have been clarified or cleaned up for improved observability during state transitions.
  • Comments/variable names have been added or clarified where appropriate.

Tests

Unit tests have been added to cover reconciliation scenarios:

  • CREATING: verifies the operator waits for creation to complete and transitions correctly to RUNNING.
  • DELETING: verifies the operator waits for deletion to complete and then re-creates the cluster.
  • DELETING (timeout): verifies the operator raises an AirflowException when the cluster remains in DELETING state and deletion is not triggered.
  • STOPPED: verifies the operator triggers cluster start logic.
  • ERROR: verifies error-state handling and deletion behavior when delete_on_error=True.

Existing tests have been updated to align with the new reconciliation flow and state handling behavior.

Backwards Compatibility

There is no intended change to the operator’s public contract. The implementation now consistently executes the previously defined reconciliation logic in non-deferrable mode.

Closes: #61947

@boring-cyborg boring-cyborg Bot added area:providers provider:google Google (including GCP) related issues labels Feb 15, 2026

@shahar1 shahar1 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

mypy currently fails :(

@SameerMesiah97

Copy link
Copy Markdown
Contributor Author

mypy currently fails :(

Yes. I will fix it.

@SameerMesiah97
SameerMesiah97 force-pushed the 61947-DataprocCreateClusterOperator-Recon-Fix branch from 613a843 to 747bf37 Compare February 15, 2026 18:06

@shahar1 shahar1 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks! I got a couple of comments.
To get this merged after applying the fixes - could you please run the system tests with your changes applied? If you're unable to do so please let me know and we'll figure it out.

Comment thread providers/google/src/airflow/providers/google/cloud/operators/dataproc.py Outdated
@SameerMesiah97
SameerMesiah97 force-pushed the 61947-DataprocCreateClusterOperator-Recon-Fix branch 2 times, most recently from de87984 to d68d5d9 Compare February 17, 2026 20:18
@SameerMesiah97

Copy link
Copy Markdown
Contributor Author

@shahar1

I have made a few changes in response to your comments. I have also ran all the system tests in:

providers\google\tests\system\google\cloud\dataproc

All 24 tests passed.

@shahar1 shahar1 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Well done, LGTM!

@VladaZakharova @MaksYermak - what do you think?

@SameerMesiah97

Copy link
Copy Markdown
Contributor Author

@VladaZakharova @MaksYermak

Tagging you here for review.

@renu-j renu-j left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the changes! I had a couple of comments during the review.

One point is related to the use of assert_called_once_with() in the test, which appears to be used twice with different arguments. Since this assertion guarantees a single call, it may lead to inconsistent expectations if multiple calls are intended. It might be clearer to use assert_has_calls() if verifying multiple calls.

I also left a small question regarding the cluster state verification after recreating the cluster in the DELETING branch.

Comment thread providers/google/tests/unit/google/cloud/operators/test_dataproc.py
…iliation runs after creation completes.

– Extract reconciliation logic into `_reconcile_cluster_state()`
– Ensure DELETING state waits for deletion and re-creates the cluster
– Ensure CREATING state is fully reconciled before returning
– Handle STOPPED state via restart path
– Raise explicit exception if cluster is not found after LRO completion
– Return reconciled cluster to avoid stale state

Update and extend unit tests to cover reconciliation scenarios in the non-deferrable path (CREATING, DELETING, STOPPED, ERROR, and timeout cases).
@SameerMesiah97
SameerMesiah97 force-pushed the 61947-DataprocCreateClusterOperator-Recon-Fix branch from d68d5d9 to 130332c Compare March 9, 2026 21:40
@SameerMesiah97

Copy link
Copy Markdown
Contributor Author

@renu-j

I appreciate you taking out the time to review the PR. I have responded to your comments. Please review.

@SameerMesiah97
SameerMesiah97 requested a review from shahar1 March 10, 2026 09:33
@potiuk
potiuk merged commit 4a8a803 into apache:main Mar 10, 2026
92 checks passed
dominikhei pushed a commit to dominikhei/airflow that referenced this pull request Mar 11, 2026
…iliation runs after creation completes. (apache#61951)

– Extract reconciliation logic into `_reconcile_cluster_state()`
– Ensure DELETING state waits for deletion and re-creates the cluster
– Ensure CREATING state is fully reconciled before returning
– Handle STOPPED state via restart path
– Raise explicit exception if cluster is not found after LRO completion
– Return reconciled cluster to avoid stale state

Update and extend unit tests to cover reconciliation scenarios in the non-deferrable path (CREATING, DELETING, STOPPED, ERROR, and timeout cases).

Co-authored-by: Sameer Mesiah <smesiah971@gmail.com>
Pyasma pushed a commit to Pyasma/airflow that referenced this pull request Mar 13, 2026
…iliation runs after creation completes. (apache#61951)

– Extract reconciliation logic into `_reconcile_cluster_state()`
– Ensure DELETING state waits for deletion and re-creates the cluster
– Ensure CREATING state is fully reconciled before returning
– Handle STOPPED state via restart path
– Raise explicit exception if cluster is not found after LRO completion
– Return reconciled cluster to avoid stale state

Update and extend unit tests to cover reconciliation scenarios in the non-deferrable path (CREATING, DELETING, STOPPED, ERROR, and timeout cases).

Co-authored-by: Sameer Mesiah <smesiah971@gmail.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

area:providers provider:google Google (including GCP) related issues

Projects

None yet

Development

Successfully merging this pull request may close these issues.

DataprocCreateClusterOperator incorrectly succeeds when cluster is deleted during creation (non-deferrable)

4 participants