Issue 1:
Apache Airflow version

Other Airflow 2 version

What happened

We have a DAG running with a cronjob schedule every 4 minutes that checks whether there is a new file present in a file system or not. If there is one available, it lists its path and passes it as a dynamic task via XCOM (one task = one file).

We have experienced a couple of times already that this task is never started, nor does it appear in the logs (not in scheduler logs, nor in AKS logs). It simply disappears and the status of the task is removed.

The DAG has following configurations:

Schedule interval: "*/4 5-22 * * 1-5"

Catchup: False

End Date: None

Max Active Runs: 1

Concurrency: 1

DAG Run Timeout: 0:10:00

What you think should happen instead

The task should be triggered and started. Instead, it gets the status removed without any logs or trace

How to reproduce

This happens in random moments, so exact steps to reproduce are hard to determine

Operating System

Airflow base docker image

Versions of Apache Airflow Providers

apache-airflow-providers-microsoft-azure: 4.2.0

Deployment

Official Apache Airflow Helm Chart

Deployment details

We are running on AKS with a postgresql backend, with the airflow helm chart of 2.3.3. Docker image is the airflow base image.

Logs are being sent to an azure file share.

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

Issue 2:
Apache Airflow version

2.3.3

What happened

If you try to backfill a DAG that uses repeated dynamic task mapping, some of the repeated mapped tasks will be marked as "removed". Any tasks downstream of the removed tasks will refuse to start:

Or, if there is nothing downstream of those tasks, the DAG Run will claim it was successful, even though some of its tasks never ran:

What you think should happen instead

All mapped tasks should run as normal during a backfill.

How to reproduce

#!/usr/bin/env python3
import datetime
import itertools
import logging
import string

from airflow.decorators import dag, task

logger = logging.getLogger(__name__)

TASK_COUNT = 25

@dag(
schedule_interval='@daily',
start_date=datetime.datetime(2022, 8, 16),
default_args={
'retries': 0,
},
)
def test_backfill_removed():
@task
def get_tasks():
return list(itertools.islice(itertools.cycle(string.ascii_letters), TASK_COUNT))

@task
def t1(arg):
logger.info(f'{arg=}')
return arg

t1.expand(arg=t1.expand(arg=get_tasks()))
# t1.expand(arg=t1.expand(arg=t1.expand(arg=get_tasks())))

dag = test_backfill_removed()

if __name__ == '__main__':
dag.cli()

airflow dags backfill test_backfill_removed -s 2022-08-01 -e 2022-08-05

Operating System

CentOS Stream 8

Versions of Apache Airflow Providers

None

Deployment

Other

Deployment details

Self-hosted + standalone

Anything else

I've reproduced this with SQLite + SequentialExecutor (standalone), Postgres + LocalExecutor (standalone), and Postgres + CeleryExecutor (production deployment, self-hosted).

I don't see any errors or warnings in the logs. In fact, even with logging_level = DEBUG, I don't see any log lines that reference the tasks that are listed as "removed".

This error is sometimes inconsistent to reproduce - if you are unlucky, you might instead get the deadlock mentioned in #25730.

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