You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "laksh-krishna-sharma (via GitHub)" <gi...@apache.org> on 2025/09/26 23:21:12 UTC
[PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
laksh-krishna-sharma opened a new pull request, #56158:
URL: https://github.com/apache/airflow/pull/56158
### Summary of changes
Closes #56153
- Deprecated `wait_policy` parameter in `EmrCreateJobFlowOperator`.
- Added `wait_for_completion` (boolean) as the standard parameter (consistent with other Amazon operators).
- Preserved backward compatibility by mapping `wait_policy` → `wait_for_completion`.
- Updated operator docstring.
### Changes made
- Refactored constructor logic to prefer `wait_for_completion`.
- Deprecated `wait_policy` with warning.
- Updated `execute` to rely on `wait_for_completion`.
- Updated operator documentation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 commented on PR #56158:
URL: https://github.com/apache/airflow/pull/56158#issuecomment-3342064952
@laksh-krishna-sharma please tag me when you're done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #56158:
URL: https://github.com/apache/airflow/pull/56158#discussion_r2384296825
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -748,8 +747,8 @@ def execute(self, context: Context) -> str | None:
job_flow_id=self._job_flow_id,
log_uri=get_log_uri(emr_client=self.hook.conn, job_flow_id=self._job_flow_id),
)
- if self.wait_policy:
- waiter_name = WAITER_POLICY_NAME_MAPPING[self.wait_policy]
+ if self.wait_for_completion:
+ waiter_name = WAITER_POLICY_NAME_MAPPING[WaitPolicy.WAIT_FOR_COMPLETION]
Review Comment:
The code hardcodes `WaitPolicy.WAIT_FOR_COMPLETION`, removing support for `WaitPolicy.WAIT_FOR_STEPS_COMPLETION` that was previously available through the `wait_policy` parameter. This is a breaking change that eliminates functionality.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -698,19 +697,19 @@ def __init__(
super().__init__(**kwargs)
self.emr_conn_id = emr_conn_id
self.job_flow_overrides = job_flow_overrides or {}
- self.wait_policy = wait_policy
+ self.wait_for_completion = wait_for_completion
self.waiter_max_attempts = waiter_max_attempts or 60
self.waiter_delay = waiter_delay or 60
self.deferrable = deferrable
- if wait_for_completion is not None:
+ if wait_policy is not None:
warnings.warn(
- "`wait_for_completion` parameter is deprecated, please use `wait_policy` instead.",
+ "`wait_policy` parameter is deprecated, please use `wait_for_completion` instead.",
AirflowProviderDeprecationWarning,
stacklevel=2,
)
# preserve previous behaviour
- self.wait_policy = WaitPolicy.WAIT_FOR_COMPLETION if wait_for_completion else None
+ self.wait_for_completion = wait_policy == WaitPolicy.WAIT_FOR_COMPLETION
Review Comment:
The backward compatibility mapping only handles `WaitPolicy.WAIT_FOR_COMPLETION`, but ignores `WaitPolicy.WAIT_FOR_STEPS_COMPLETION`. Users with `wait_policy=WaitPolicy.WAIT_FOR_STEPS_COMPLETION` will have their configuration silently changed to `wait_for_completion=False`, losing the steps completion functionality.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
Posted by "laksh-krishna-sharma (via GitHub)" <gi...@apache.org>.
laksh-krishna-sharma commented on PR #56158:
URL: https://github.com/apache/airflow/pull/56158#issuecomment-3342008050
yes @shahar1 i am reviewing the comments
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
Posted by "laksh-krishna-sharma (via GitHub)" <gi...@apache.org>.
laksh-krishna-sharma commented on PR #56158:
URL: https://github.com/apache/airflow/pull/56158#issuecomment-3341922188
sure @shahar1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck merged PR #56158:
URL: https://github.com/apache/airflow/pull/56158
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
Posted by "laksh-krishna-sharma (via GitHub)" <gi...@apache.org>.
laksh-krishna-sharma commented on PR #56158:
URL: https://github.com/apache/airflow/pull/56158#issuecomment-3342010355
@shahar1 the copilot comments make sense. I’m working on resolving these.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #56158:
URL: https://github.com/apache/airflow/pull/56158#discussion_r2384326798
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -748,8 +751,11 @@ def execute(self, context: Context) -> str | None:
job_flow_id=self._job_flow_id,
log_uri=get_log_uri(emr_client=self.hook.conn, job_flow_id=self._job_flow_id),
)
- if self.wait_policy:
- waiter_name = WAITER_POLICY_NAME_MAPPING[self.wait_policy]
+ if self.wait_for_completion:
+ if self.wait_policy:
+ waiter_name = WAITER_POLICY_NAME_MAPPING[self.wait_policy]
+ else:
+ waiter_name = WAITER_POLICY_NAME_MAPPING[WaitPolicy.WAIT_FOR_COMPLETION]
Review Comment:
The logic is incorrect. When `wait_policy` is deprecated and mapped to `wait_for_completion`, the code should not still check `self.wait_policy` inside the `wait_for_completion` block. This could cause issues when both parameters are provided or when `wait_policy` is None but `wait_for_completion` is True.
```suggestion
waiter_name = WAITER_POLICY_NAME_MAPPING[WaitPolicy.WAIT_FOR_COMPLETION]
```
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -698,19 +697,23 @@ def __init__(
super().__init__(**kwargs)
self.emr_conn_id = emr_conn_id
self.job_flow_overrides = job_flow_overrides or {}
- self.wait_policy = wait_policy
+ self.wait_for_completion = wait_for_completion
self.waiter_max_attempts = waiter_max_attempts or 60
self.waiter_delay = waiter_delay or 60
self.deferrable = deferrable
+ self.wait_policy = wait_policy
- if wait_for_completion is not None:
+ if wait_policy is not None:
warnings.warn(
- "`wait_for_completion` parameter is deprecated, please use `wait_policy` instead.",
+ "`wait_policy` parameter is deprecated, please use `wait_for_completion` instead.",
AirflowProviderDeprecationWarning,
stacklevel=2,
)
# preserve previous behaviour
- self.wait_policy = WaitPolicy.WAIT_FOR_COMPLETION if wait_for_completion else None
+ self.wait_for_completion = wait_policy in (
+ WaitPolicy.WAIT_FOR_COMPLETION,
+ WaitPolicy.WAIT_FOR_STEPS_COMPLETION,
+ )
Review Comment:
This mapping loses important information. Both `WAIT_FOR_COMPLETION` and `WAIT_FOR_STEPS_COMPLETION` are mapped to `True`, but they have different behaviors in the original wait_policy logic. The `WAIT_FOR_STEPS_COMPLETION` behavior is now lost when using the deprecated parameter.
```suggestion
self.wait_policy = wait_policy
if wait_policy == WaitPolicy.WAIT_FOR_COMPLETION:
self.wait_for_completion = True
elif wait_policy == WaitPolicy.WAIT_FOR_STEPS_COMPLETION:
self.wait_for_completion = "steps"
else:
self.wait_for_completion = False
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
Posted by "laksh-krishna-sharma (via GitHub)" <gi...@apache.org>.
laksh-krishna-sharma commented on PR #56158:
URL: https://github.com/apache/airflow/pull/56158#issuecomment-3344261952
@shahar1 please review the changes and let me know if i have to refactor
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #56158:
URL: https://github.com/apache/airflow/pull/56158#discussion_r2384319904
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -748,8 +751,11 @@ def execute(self, context: Context) -> str | None:
job_flow_id=self._job_flow_id,
log_uri=get_log_uri(emr_client=self.hook.conn, job_flow_id=self._job_flow_id),
)
- if self.wait_policy:
- waiter_name = WAITER_POLICY_NAME_MAPPING[self.wait_policy]
+ if self.wait_for_completion:
+ if hasattr(self, "wait_policy") and self.wait_policy:
Review Comment:
The `hasattr(self, "wait_policy")` check is unnecessary since `self.wait_policy` is always set in the constructor when `wait_policy` parameter is provided. Consider simplifying this to just check `if self.wait_policy:` since the attribute will exist when needed.
```suggestion
if self.wait_policy:
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 commented on PR #56158:
URL: https://github.com/apache/airflow/pull/56158#issuecomment-3342007310
@laksh-krishna-sharma could you please review Copilot's comments and check if they make sense?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #56158:
URL: https://github.com/apache/airflow/pull/56158#discussion_r2384375000
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -698,19 +697,22 @@ def __init__(
super().__init__(**kwargs)
self.emr_conn_id = emr_conn_id
self.job_flow_overrides = job_flow_overrides or {}
- self.wait_policy = wait_policy
+ self.wait_for_completion = wait_for_completion
self.waiter_max_attempts = waiter_max_attempts or 60
self.waiter_delay = waiter_delay or 60
self.deferrable = deferrable
- if wait_for_completion is not None:
+ if wait_policy is not None:
warnings.warn(
- "`wait_for_completion` parameter is deprecated, please use `wait_policy` instead.",
+ "`wait_policy` parameter is deprecated, please use `wait_for_completion` instead.",
AirflowProviderDeprecationWarning,
stacklevel=2,
)
- # preserve previous behaviour
- self.wait_policy = WaitPolicy.WAIT_FOR_COMPLETION if wait_for_completion else None
+ self.wait_policy = wait_policy
+ self.wait_for_completion = wait_policy in (
+ WaitPolicy.WAIT_FOR_COMPLETION,
+ WaitPolicy.WAIT_FOR_STEPS_COMPLETION,
+ )
Review Comment:
The `wait_policy` is still being stored but never used in the execute method. Consider removing this assignment since the execute method now only uses `wait_for_completion`, or preserve the original waiter selection logic to maintain full backward compatibility.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 commented on code in PR #56158:
URL: https://github.com/apache/airflow/pull/56158#discussion_r2386297379
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -698,19 +697,22 @@ def __init__(
super().__init__(**kwargs)
self.emr_conn_id = emr_conn_id
self.job_flow_overrides = job_flow_overrides or {}
- self.wait_policy = wait_policy
+ self.wait_for_completion = wait_for_completion
self.waiter_max_attempts = waiter_max_attempts or 60
self.waiter_delay = waiter_delay or 60
self.deferrable = deferrable
- if wait_for_completion is not None:
+ if wait_policy is not None:
warnings.warn(
- "`wait_for_completion` parameter is deprecated, please use `wait_policy` instead.",
+ "`wait_policy` parameter is deprecated, please use `wait_for_completion` instead.",
AirflowProviderDeprecationWarning,
stacklevel=2,
)
- # preserve previous behaviour
- self.wait_policy = WaitPolicy.WAIT_FOR_COMPLETION if wait_for_completion else None
+ self.wait_policy = wait_policy
+ self.wait_for_completion = wait_policy in (
+ WaitPolicy.WAIT_FOR_COMPLETION,
+ WaitPolicy.WAIT_FOR_STEPS_COMPLETION,
+ )
Review Comment:
@laksh-krishna-sharma please try to polish this one.
After that I think that we're good :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #56158:
URL: https://github.com/apache/airflow/pull/56158#discussion_r2386336707
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -748,8 +758,8 @@ def execute(self, context: Context) -> str | None:
job_flow_id=self._job_flow_id,
log_uri=get_log_uri(emr_client=self.hook.conn, job_flow_id=self._job_flow_id),
)
- if self.wait_policy:
- waiter_name = WAITER_POLICY_NAME_MAPPING[self.wait_policy]
+ if self.wait_for_completion:
+ waiter_name = WAITER_POLICY_NAME_MAPPING[WaitPolicy.WAIT_FOR_COMPLETION]
Review Comment:
The hardcoded waiter policy loses functionality when `wait_policy` was `WaitPolicy.WAIT_FOR_STEPS_COMPLETION`. The original behavior should be preserved by storing the original wait policy and using it here, or by extending the boolean parameter to support the steps completion option.
```suggestion
waiter_name = WAITER_POLICY_NAME_MAPPING[self.wait_policy]
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
Posted by "laksh-krishna-sharma (via GitHub)" <gi...@apache.org>.
laksh-krishna-sharma commented on PR #56158:
URL: https://github.com/apache/airflow/pull/56158#issuecomment-3352020475
@vincbeck @shahar1 please review the changes and let me know if i have to refactor
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #56158:
URL: https://github.com/apache/airflow/pull/56158#discussion_r2387919690
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -698,19 +697,30 @@ def __init__(
super().__init__(**kwargs)
self.emr_conn_id = emr_conn_id
self.job_flow_overrides = job_flow_overrides or {}
- self.wait_policy = wait_policy
+ self.wait_for_completion = wait_for_completion
self.waiter_max_attempts = waiter_max_attempts or 60
self.waiter_delay = waiter_delay or 60
self.deferrable = deferrable
- if wait_for_completion is not None:
+ if wait_policy is not None:
warnings.warn(
- "`wait_for_completion` parameter is deprecated, please use `wait_policy` instead.",
+ "`wait_policy` parameter is deprecated and will be removed in a future release; "
+ "please use `wait_for_completion` (bool) instead.",
AirflowProviderDeprecationWarning,
stacklevel=2,
)
- # preserve previous behaviour
- self.wait_policy = WaitPolicy.WAIT_FOR_COMPLETION if wait_for_completion else None
+ self.wait_policy = wait_policy
Review Comment:
I do not think we need to save it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
Posted by "laksh-krishna-sharma (via GitHub)" <gi...@apache.org>.
laksh-krishna-sharma commented on PR #56158:
URL: https://github.com/apache/airflow/pull/56158#issuecomment-3342039839
@shahar1 Can I mark the Copilot comments as resolved?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor: deprecate wait_policy in EmrCreateJobFlowOperator in favor of wait_for_completion [airflow]
Posted by "laksh-krishna-sharma (via GitHub)" <gi...@apache.org>.
laksh-krishna-sharma commented on PR #56158:
URL: https://github.com/apache/airflow/pull/56158#issuecomment-3342094367
@shahar1 please review the changes and let me know if i have to refactor
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org