Each scheduler may process same zombie tasks in HA mode

Apache Airflow version

main (development)

What happened

Lines 1447 to 1483 in f40ec8d

    
               def _find_zombies(self, session: Session) -> None: 
        
                   """ 
        
                   Find zombie task instances, which are tasks haven't heartbeated for too long 
        
                   or have a no-longer-running LocalTaskJob, and create a TaskCallbackRequest 
        
                   to be handled by the DAG processor. 
        
                   """ 
        
                   self.log.debug("Finding 'running' jobs without a recent heartbeat") 
        
                   limit_dttm = timezone.utcnow() - timedelta(seconds=self._zombie_threshold_secs) 
        
                   zombies = ( 
        
                       session.query(TaskInstance, DagModel.fileloc) 
        
                       .with_hint(TI, 'USE INDEX (ti_state)', dialect_name='mysql') 
        
                       .join(LocalTaskJob, TaskInstance.job_id == LocalTaskJob.id) 
        
                       .join(DagModel, TaskInstance.dag_id == DagModel.dag_id) 
        
                       .filter(TaskInstance.state == TaskInstanceState.RUNNING) 
        
                       .filter( 
        
                           or_( 
        
                               LocalTaskJob.state != State.RUNNING, 
        
                               LocalTaskJob.latest_heartbeat < limit_dttm, 
        
                           ) 
        
                       ) 
        
                       .filter(TaskInstance.queued_by_job_id == self.id) 
        
                       .all() 
        
                   ) 
        
                   if zombies: 
        
                       self.log.warning("Failing (%s) jobs without heartbeat after %s", len(zombies), limit_dttm) 
        
                   for ti, file_loc in zombies: 
        
                       request = TaskCallbackRequest( 
        
                           full_filepath=file_loc, 
        
                           simple_task_instance=SimpleTaskInstance.from_ti(ti), 
        
                           msg=f"Detected {ti} as zombie", 
        
                       ) 
        
                       self.log.error("Detected zombie job: %s", request) 
        
                       self.executor.send_callback(request) 
        
                       Stats.incr('zombies_killed')

_find_zombies should also support HA. Each scheduler tries to find all zombie tasks. When there are more than one schedulers, it might cause issues.

What you think should happen instead

No response

How to reproduce

No response

Operating System

Apple M1 Max, version: 12.2

Versions of Apache Airflow Providers

No response

Deployment

Other

Deployment details

NA

Anything else

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct

Can you try 2.3.4RC1 #25846?
#24906 probably fixed it?

Yep. This is fixed in #24906

@eladkal

nice, thanks @eladkal and @potiuk

Navigation Menu

Repository navigation

Each scheduler may process same zombie tasks in HA mode #25843

Apache Airflow version

What happened

What you think should happen instead

How to reproduce

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

Anything else

Are you willing to submit PR?

Code of Conduct

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	def _find_zombies(self, session: Session) -> None:
	"""
	Find zombie task instances, which are tasks haven't heartbeated for too long
	or have a no-longer-running LocalTaskJob, and create a TaskCallbackRequest
	to be handled by the DAG processor.
	"""
	self.log.debug("Finding 'running' jobs without a recent heartbeat")
	limit_dttm = timezone.utcnow() - timedelta(seconds=self._zombie_threshold_secs)

	zombies = (
	session.query(TaskInstance, DagModel.fileloc)
	.with_hint(TI, 'USE INDEX (ti_state)', dialect_name='mysql')
	.join(LocalTaskJob, TaskInstance.job_id == LocalTaskJob.id)
	.join(DagModel, TaskInstance.dag_id == DagModel.dag_id)
	.filter(TaskInstance.state == TaskInstanceState.RUNNING)
	.filter(
	or_(
	LocalTaskJob.state != State.RUNNING,
	LocalTaskJob.latest_heartbeat < limit_dttm,
	)
	)
	.filter(TaskInstance.queued_by_job_id == self.id)
	.all()
	)

	if zombies:
	self.log.warning("Failing (%s) jobs without heartbeat after %s", len(zombies), limit_dttm)

	for ti, file_loc in zombies:
	request = TaskCallbackRequest(
	full_filepath=file_loc,
	simple_task_instance=SimpleTaskInstance.from_ti(ti),
	msg=f"Detected {ti} as zombie",
	)
	self.log.error("Detected zombie job: %s", request)
	self.executor.send_callback(request)
	Stats.incr('zombies_killed')

Each scheduler may process same zombie tasks in HA mode #25843

Description

Apache Airflow version

What happened

What you think should happen instead

How to reproduce

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

Anything else

Are you willing to submit PR?

Code of Conduct

Activity

eladkal commented on Aug 20, 2022

potiuk commented on Aug 20, 2022

pingzh commented on Aug 22, 2022

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions