Depreacate EmrContainerSensorAsync and EmrStepSensorAsync

[Lee-W]

What's changed

Deprecate EmrContainerSensorAsync and EmrStepSensorAsync and fallback to EmrContainerSensor and EmrStepSensor with deferrable=True

Why this change

The logic of EmrContainerSensorAsync and EmrStepSensorAsync has been contributed back to OSS airflow as part of EmrContainerSensor and EmrStepSensor

EmrContainerSensorAsync

astronomer-providers/astronomer/providers/amazon/aws/sensors/emr.py

Lines 38 to 65 in b5a9417

	def execute(self, context: Context) -> None:
	"""Defers trigger class to poll for state of the job run until it reaches a failure state or success state"""
	if not poke(self, context):
	self.defer(
	timeout=timedelta(seconds=self.timeout),
	trigger=EmrContainerSensorTrigger(
	virtual_cluster_id=self.virtual_cluster_id,
	job_id=self.job_id,
	max_tries=self.max_retries,
	aws_conn_id=self.aws_conn_id,
	poll_interval=self.poll_interval,
	),
	method_name="execute_complete",
	)
	# Ignoring the override type check because the parent class specifies "context: Any" but specifying it as
	# "context: Context" is accurate as it's more specific.
	def execute_complete(self, context: Context, event: dict[str, str]) -> None: # type: ignore[override]
	"""
	Callback for when the trigger fires - returns immediately.
	Relies on trigger to throw an exception, otherwise it assumes execution was
	successful.
	"""
	if event:
	if event["status"] == "error":
	raise_error_or_skip_exception(self.soft_fail, event["message"])
	self.log.info(event["message"])
	return None

EmrContainerSensor

https://github.com/apache/airflow/blob/999b70178a1f5d891fd2c88af4831a4ba4c2cbc9/airflow/providers/amazon/aws/sensors/emr.py#L318-L346

EmrStepSensorAsync

astronomer-providers/astronomer/providers/amazon/aws/sensors/emr.py

Lines 87 to 113 in b5a9417

	def execute(self, context: Context) -> None:
	"""Deferred and give control to trigger"""
	if not poke(self, context):
	self.defer(
	timeout=timedelta(seconds=self.timeout),
	trigger=EmrStepSensorTrigger(
	job_flow_id=self.job_flow_id,
	step_id=self.step_id,
	target_states=self.target_states,
	failed_states=self.failed_states,
	aws_conn_id=self.aws_conn_id,
	poke_interval=self.poke_interval,
	),
	method_name="execute_complete",
	)
	def execute_complete(self, context: Context, event: dict[str, Any]) -> None: # type: ignore[override]
	"""
	Callback for when the trigger fires - returns immediately.
	Relies on trigger to throw an exception, otherwise it assumes execution was
	successful.
	"""
	if event:
	if event["status"] == "error":
	raise_error_or_skip_exception(self.soft_fail, event["message"])
	self.log.info(event.get("message"))
	self.log.info("%s completed successfully.", self.job_flow_id)

EmrStepSensor

https://github.com/apache/airflow/blob/999b70178a1f5d891fd2c88af4831a4ba4c2cbc9/airflow/providers/amazon/aws/sensors/emr.py#L644-L668

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Comparison is base (6781996) 98.57% compared to head (148fba4) 98.55%.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #1390      +/-   ##
==========================================
- Coverage   98.57%   98.55%   -0.02%     
==========================================
  Files          91       91              
  Lines        5460     5403      -57     
==========================================
- Hits         5382     5325      -57     
  Misses         78       78              

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[phanikumv]

what about the integration DAGs that might be using these classes, are we planning to remove them at a later point of time?

[phanikumv]

Please add description about why you are doing this change

[pankajkoti]

We will also need to remove corresponding trigger modules, no?

[Lee-W]

what about the integration DAGs that might be using this class, are we planning to remove them at a later point of time?

Yes, I think we should keep it for a while and remove it after we deprecate the whole repo.

Please add description about why you are doing this change

Sure. let me reformat and reword it to make it easier to read.

[Lee-W]

We will also need to remove corresponding trigger modules, no?

Yep, we should remove it as well. Thanks for reminding me!