ShortCircuitOperator not skipping tasks

[matthieucx]

Hello! I'm having a hard time figuring out an issue I have.

I have a Dag responsible for updating an Optuna study. Training jobs are running during the study update, and as a consequence the study won't be finalized if the last training job finishes while the dag is running.

Thus, if all training jobs have been launched, I want to wait for them using an External Task Sensor. If not, I want to skip the wait.

Here are my tasks dependencies:

    metadata = get_metadata()
    should_wait_bool = update_study.partial().expand(metadata=metadata)
    should_wait_training_job(should_wait_bool) >> wait_training_job >> trigger_study_update()

My ShortCircuitOperator:

    @task.short_circuit()
    def should_wait_training_job(need_wait_training_job):
        return need_wait_training_job

wait_training_job is an ExternalTaskSensor.

Here are the logs for the short-circuit task:

[2023-04-24, 15:02:58 UTC] {taskinstance.py:1300} INFO - Executing <Task(_ShortCircuitDecoratedOperator): should_wait_training_job> on 2023-04-24 14:54:07.730299+00:00
[2023-04-24, 15:02:58 UTC] {standard_task_runner.py:55} INFO - Started process 62133 to run task
[2023-04-24, 15:02:58 UTC] {standard_task_runner.py:82} INFO - Running: ['***', 'tasks', 'run', 'update_study', 'should_wait_training_job', 'dataset_triggered__2023-04-24T14:54:07.730299+00:00', '--job-id', '8822', '--raw', '--subdir', 'DAGS_FOLDER/dags/update_study.py', '--cfg-path', '/tmp/tmpba8eg5n7']
[2023-04-24, 15:02:58 UTC] {standard_task_runner.py:83} INFO - Job 8822: Subtask should_wait_training_job
[2023-04-24, 15:02:58 UTC] {task_command.py:388} INFO - Running <TaskInstance: update_study.should_wait_training_job dataset_triggered__2023-04-24T14:54:07.730299+00:00 [running]> on host 2108dab88f78
[2023-04-24, 15:02:58 UTC] {taskinstance.py:1507} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=***
AIRFLOW_CTX_DAG_ID=update_study
AIRFLOW_CTX_TASK_ID=should_wait_training_job
AIRFLOW_CTX_EXECUTION_DATE=2023-04-24T14:54:07.730299+00:00
AIRFLOW_CTX_TRY_NUMBER=2
AIRFLOW_CTX_DAG_RUN_ID=dataset_triggered__2023-04-24T14:54:07.730299+00:00
[2023-04-24, 15:02:58 UTC] {python.py:177} INFO - Done. Returned value was: [False]
[2023-04-24, 15:02:58 UTC] {python.py:248} INFO - Condition result is [False]
[2023-04-24, 15:02:58 UTC] {python.py:251} INFO - Proceeding with downstream tasks...
[2023-04-24, 15:02:58 UTC] {xcom.py:225} WARNING - Coercing mapped lazy proxy return value from task should_wait_training_job (DAG update_study, run dataset_triggered__2023-04-24T14:54:07.730299+00:00) to list, which may degrade performance. Review resource requirements for this operation, and call list() to suppress this message. See Dynamic Task Mapping documentation for more information about lazy proxy objects.
[2023-04-24, 15:02:58 UTC] {taskinstance.py:1318} INFO - Marking task as SUCCESS. dag_id=update_study, task_id=should_wait_training_job, execution_date=20230424T145407, start_date=20230424T150258, end_date=20230424T150258
[2023-04-24, 15:02:59 UTC] {local_task_job.py:208} INFO - Task exited with return code 0
[2023-04-24, 15:02:59 UTC] {taskinstance.py:2578} INFO - 0 downstream tasks scheduled from follow-on schedule check

If anyone has an idea of what's happening that would help me greatly !

[Sighery]

Hey, I'm new to Airflow, and haven't used it much yet. We are just getting started with it. I think, and also judging by the documentation for the ShortCircuitOperator, it's considering the result as True, and that's why it continues. From the documentation:

If the returned result is False or a falsy value, the pipeline will be short-circuited. Downstream tasks will be marked with a state of “skipped” based on the short-circuiting mode configured. If the returned result is True or a truthy value, downstream tasks proceed as normal and an XCom of the returned result is pushed.

[False] is a truthy value in Python, since it's not an empty array. If you want the result to be falsy, you'll have to either unpack it so it's just False, or turn it into an empty array [], which would also be considered falsy. You can easily check what equals truthy/falsy by opening a Python interpreter and running bool([False]), or bool([]), or any such condition passed to the bool function. Here documentation for it.

[matthieucx]

Thanks mate! I believe you're right. Honestly I should have picked up that on my own, but I think I realized I was doing something convoluted in my DAG and dropped that implementation altogether

Either way, thanks for answering and good luck in learning Airflow!