You are viewing a plain text version of this content. The canonical link for it is here.

Posted to commits@airflow.apache.org by "JosephABC (via GitHub)" <gi...@apache.org> on 2024/02/03 07:43:15 UTC

[I] Subsequent Dag Run mapped task not scheduled due to removed mapped task in previous Dag Run [airflow]

JosephABC opened a new issue, #37154:
URL: https://github.com/apache/airflow/issues/37154

   ### Apache Airflow version
   
   Other Airflow 2 version (please specify below)
   
   ### If "Other Airflow 2 version" selected, which one?
   
   2.7.3
   
   ### What happened?
   
   ### **Dag Information**
   
   - `generate_records_task` will generate a list of integers based on attempt
     - 1 less record per attempt (4 -> 3 -> 2)
   
   - The list of records returned is dynamic and will serve as input to the mapped task operator to generate the intended number of mapped tasks to run on each record
   
   - `depends_on_past` is true for this DAG
   
   - We are not using task flow decorators
   - Business uses for this pattern : Identify cities/regions for which we need to run pipeline dynamically. The source of truth for this can change in real time.
   
   <img width="500" alt="dag_graph" src="https://github.com/apache/airflow/assets/25327813/1e2c2a61-2525-4265-bfc2-3aefe03fbfcc">
   
   
   
   ### **Previous Dag Run**
   
   - When we perform actions like clear dag or clear task and downstream, `generate_records_task` task will generate different number of records for each attempt
   - When a mapped task like `mapped_task_instance` was executed the second time for the same dag run, it would mark one of the mapped tasks as `removed` state. Whole mapped task was marked as success
   
   ![previous_dag_run_attempt_1](https://github.com/apache/airflow/assets/25327813/7d5ca12a-3fb6-4eaf-8bf2-13b0c698f4d7)
   
   ![previous_dag_run_attempt_2](https://github.com/apache/airflow/assets/25327813/2570346b-98ad-421f-9568-7d07b56dc4b7)
   
   ![previous_dag_run](https://github.com/apache/airflow/assets/25327813/13ae030d-b7aa-4171-92fe-e761add72cc1)
   
   ### **Subsequent Dag Run**
   
   - We tried 2 scenarios for subsequent Dag Run
     - Clear Dag Run
     - Delete Dag Run (similar to having a new Dag Run)
   - In both senarios, the `mapped_task_instance` task did not schedule and dag run was stuck in `running` state
     - For Clear Dag Run: We can see that there are 3 mapped with no_status and 1 with removed. 
     - For Delete Dag Run: No mapped tasks were generated
   
   ![subsequent_dag_run_clear](https://github.com/apache/airflow/assets/25327813/b3d85005-c1ef-4d05-a8ae-c12050d990a6)
   
     - For Delete Dag Run: No mapped tasks were generated
   
   ![subsequent_dag_run_delete](https://github.com/apache/airflow/assets/25327813/ce1368f9-e790-48b3-ba78-deacfef631a3)
   
   ### **Temporary Resolution**
   When we delete the `removed` mapped task for `mapped_task_instance` in the previous dag run, scheduling resumed for `mapped_task_instance` for current dag run
   
   
   
   ### What you think should happen instead?
   
   When `depends_on_past` is set to True, mapped task in subsequent Dag Run should only depend on the overall status of the mapped task in the previous Dag Run.
   
   Mapped task in removed state may be an intended outcome from operational changes leading to rerun of the previous Dag Run.
   
   ### How to reproduce
   
   DAG Code for reproducing error in this word document
   
   [Uploading Removed Mapped Task issue in Airflow 2.7.3 (Airflow).docx…]()
   
   
   ### Operating System
   
   Debian GNU/Linux 11 (bullseye)
   
   ### Versions of Apache Airflow Providers
   
   _No response_
   
   ### Deployment
   
   Other Docker-based deployment
   
   ### Deployment details
   
   _No response_
   
   ### Anything else?
   
   _No response_
   
   ### Are you willing to submit PR?
   
   - [X] Yes I am willing to submit a PR!
   
   ### Code of Conduct
   
   - [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
   


-- 
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.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] Subsequent Dag Run mapped task not scheduled due to removed mapped task in previous Dag Run [airflow]

Posted by "arunravimv (via GitHub)" <gi...@apache.org>.

arunravimv commented on issue #37154:
URL: https://github.com/apache/airflow/issues/37154#issuecomment-1925372542

   Adding some additional info, scheduler logs keeps logging. We run scheduler in HA with 2 replica and our backed store in MYSQL from AWS RDS.
   ```
   [2024-02-03T15:14:13.171+0000] {scheduler_job_runner.py:1433} INFO - DAG mapped_task_issue scheduling was skipped, probably because the DAG record was locked
   ```
   
   We are new to Airflow Source code, but we saw this [possible area](https://github.com/apache/airflow/blob/d987bce144ac8e42bd269e1971d0ad4f15dcff95/airflow/ti_deps/deps/prev_dagrun_dep.py#L83) where the problem might be originating from. The _count_unsuccessful_tis func in PrevDagrunDep Class only considers SKIPPED and SUCCESS as successful state, but I believe in this case REMOVED is also a valid successful state. 
   
   However we don't know why the lasted dag run is not getting marked as failed here with [unmet dependency error here](https://github.com/apache/airflow/blob/d987bce144ac8e42bd269e1971d0ad4f15dcff95/airflow/ti_deps/deps/prev_dagrun_dep.py#L181) and why is it in stuck state


-- 
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: [I] Subsequent Dag Run mapped task not scheduled due to removed mapped task in previous Dag Run [airflow]

Posted by "JosephABC (via GitHub)" <gi...@apache.org>.

JosephABC commented on issue #37154:
URL: https://github.com/apache/airflow/issues/37154#issuecomment-2151677374

   @nathadfield , we were able to replicate the issue with the following code on Airflow v2.9.1
   Following code is slightly modified from the one above. It should create the scenario automatically, will just need to enable the dag and have at least 2 dag runs.
   
   ```python
   from airflow.models import DAG
   from datetime import date, datetime, timedelta
   
   from airflow.operators.python_operator import PythonOperator
   
   default_args = {"owner": "joseph.ang", "depends_on_past": True, "retries": 1, "retry_delay": timedelta(minutes=1), "max_active_runs":1}
   
   dag = DAG(
       "mapped_task_issue",
       description="Testing mapped task issue in Airflow v2.9.1",
       schedule_interval="0 0 * * *",
       catchup=True,
       start_date=datetime(2024, 6, 4),
       default_args=default_args,
   )
   
   
   def generate_records(**context):
       try_number = context['task_instance'].try_number
       try_number = 5 - try_number
   
       generated_list = [ {'i': i} for i in range(try_number)]
   
       print(f'Generated list: {generated_list}')
   
       return generated_list
   
   def mapped_task_method(i: int):
       print(i)
       if i == 3:
           exit(1)
   
   
   generate_records_task = PythonOperator(
       task_id='generate_records_task',
       python_callable=generate_records,
       provide_context=True,
       dag=dag,
   )
   
   mapped_task = PythonOperator.partial(
       task_id='mapped_task_instance',
       python_callable=mapped_task_method,
       dag=dag,
   ).expand(op_kwargs=generate_records_task.output)
   ```
   <img width="1896" alt="image" src="https://github.com/apache/airflow/assets/25327813/201e2d02-554b-47f7-845d-0b36802e17cd">
   


-- 
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: [I] Subsequent Dag Run mapped task not scheduled due to removed mapped task in previous Dag Run [airflow]

Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.

boring-cyborg[bot] commented on issue #37154:
URL: https://github.com/apache/airflow/issues/37154#issuecomment-1925200997

   Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.
   


-- 
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