In Airflow, we have a series of tasks. But to simplify let us consider there are 2 tasks. The first task lists files from an S3 bucket and the second task picks those files to process.
list_files_task = list_files()
transform_files_task = transform_file.expand(source_s3_key=list_files_task)
list_files_task >> transform_files_task
Very rarely (once in a week or so, it is running at 5 min cadence) what is happening is that the first task completes in a second and the second task gets picked like 12 hours later. What could be the reason? Our tasks are not super complicated and nothing seem off about our code.
