[GitHub] [airflow] quoc-t-le opened a new issue #15407: ShortCircuitOperator to short circuit downstream of TaskGroup only

[GitBox]

quoc-t-le opened a new issue #15407:
URL: https://github.com/apache/airflow/issues/15407


   **Description**
   
   I am trying to convert a SubDag to a TaskGroup which uses the 
ShortCircuitOperator.  Originally, the short circuit will trip and move on to 
next SubDag.  I cannot get the same behavior with TaskGroup.
   
   What do you want to happen?
   
   If used ShortCircuitOperator within a TaskGroup, I would expect it to just 
short circuit the downstreams within the taskgroup.
   
   Here is the code I am trying to do.  I would expect T1 to be short circuited 
and execute T2:
   
   `from airflow import DAG
   from datetime import datetime
   from airflow.operators.dummy_operator import DummyOperator
   from airflow.operators.python_operator import ShortCircuitOperator, 
PythonOperator
   from airflow.utils.task_group import TaskGroup
   
   default_args = {
       'owner': 'airflow',
       'retries': 3,
       'depends_on_past': False,
   }
   
   def short_circuit (*arg, **kwargs):
       return False
   
   with DAG ("short-circuit",
             catchup=True,
             default_args=default_args,
             schedule_interval='@daily',
             description='Aggregates and pulls down data for API endpoints that 
use analytics',
             start_date=datetime.strptime('04/14/2021', '%m/%d/%Y'),
             max_active_runs=1
   ) as dag:
       t0 = DummyOperator(task_id='start')
       with TaskGroup('t1') as t1:
           s1 = ShortCircuitOperator(
               task_id='short_circuit',
               python_callable=short_circuit
           )
           s2 = DummyOperator(task_id='t1s2')
           s3 = DummyOperator(task_id='t1s3')
           s4 = DummyOperator(task_id='t1s4')
           s5 = DummyOperator(task_id='t1s5')
           s6 = DummyOperator(task_id='t1s6')
           s7 = DummyOperator(task_id='t1s7')
           s8 = DummyOperator(task_id='t1s8')
           s9 = DummyOperator(task_id='t1s9')
           s1 >> s2 >> s3 >> s4
           s4 >> s5
           s5 >> s6 >> s7
           s7 >> s8 >> s9
   
   
       with TaskGroup('t2') as t2:
           s1 = ShortCircuitOperator(
                       task_id='short_circuit',
                       python_callable=short_circuit,
                       op_kwargs={"category": "mobile"}
                   )
           s2 = DummyOperator(task_id='t1s2')
           s3 = DummyOperator(task_id='t1s3')
           s4 = DummyOperator(task_id='t1s4')
           s5 = DummyOperator(task_id='t1s5')
           s6 = DummyOperator(task_id='t1s6')
           s7 = DummyOperator(task_id='t1s7')
           s8 = DummyOperator(task_id='t1s8')
           s9 = DummyOperator(task_id='t1s9')
           s1 >> s2 >> s3 >> s4
           s4 >> s5
           s5 >> s6 >> s7
           s7 >> s8 >> s9
   
       t0 >> t1 >> t2
   `
   
   
   


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

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