Issue 1:
Apache Airflow version: 2.0.1

Kubernetes version (if you are using kubernetes) (use kubectl version):1.11

Environment:

Cloud provider or hardware configuration: OpenShift 3.11

OS (e.g. from /etc/os-release): RHEL Server

Kernel (e.g. uname -a): Linux

Install tools: Airflow

Others:

What happened:

HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"pods "sales-trend-predict.181186911d0344eda3f3ff7d446f7d62" is forbidden: [maximum memory usage per Pod is 8Gi, but limit is 9663676416., minimum cpu usage per Container is 400m, but request is 1m., cpu max limit to request ratio per Container is 5, but provided ratio is 1000.000000.]","reason":"Forbidden","details":{"name":"sales-trend-predict.181186911d0344eda3f3ff7d446f7d62","kind":"pods"},"code":403}

What you expected to happen:

For some enterprise kubernetes/openshift cluster, the admin will set a cpu/memory limit on container's resource, eg. minimum 400m cpu core, maximum 8Gi memory etc. However, for airflow's auto generated sidecar container for xcom 'airflow-xcom-sidecar' will fail this requirement as the error shown above, since the cpu core request is set to '1m' currently, (https://github.com/apache/airflow/blob/master/airflow/kubernetes/pod_generator.py#L60), as a result, the sidecar container won't be able to start up. Therefore, we need to have the capability to specify the xcom sidecar resource on kubernetesPodOperator as a parameter.

How to reproduce it:

This error only exist in certain kubernetes cluster that have a restriction on container's resource. Therefore, the error can be reproduced by manually setting a resource restriction on admin's level (eg. cpu core request 400m minimum, etc.). Then use KubernetesPodOperator with xcom enabled, you will be able to reproduce the error above.

Anything else we need to know:

N/A

Issue 2:
Apache Airflow version

2.1.4 (latest released)

Operating System

Debian GNU/Linux 10 (buster)

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==2.2.0

apache-airflow-providers-celery==2.0.0

apache-airflow-providers-cncf-kubernetes==2.0.2

apache-airflow-providers-docker==2.1.1

apache-airflow-providers-elasticsearch==2.0.3

apache-airflow-providers-ftp==2.0.1

apache-airflow-providers-google==5.1.0

apache-airflow-providers-grpc==2.0.1

apache-airflow-providers-hashicorp==2.1.0

apache-airflow-providers-http==2.0.1

apache-airflow-providers-imap==2.0.1

apache-airflow-providers-microsoft-azure==3.1.1

apache-airflow-providers-mysql==2.1.1

apache-airflow-providers-postgres==2.2.0

apache-airflow-providers-redis==2.0.1

apache-airflow-providers-sendgrid==2.0.1

apache-airflow-providers-sftp==2.1.1

apache-airflow-providers-slack==4.0.1

apache-airflow-providers-sqlite==2.0.1

apache-airflow-providers-ssh==2.1.1

Deployment

Docker-Compose

Deployment details

dockerfile from apache/airflow:2.1.4-python3.7

EXECUTOR=LocalExecutor

What happened

After successfully running a SUBDAG, clearing it (including downstream+recursive) doesn't trigger the inner tasks. Instead, the subdag is marked successful and the inner tasks all stay cleared and are not re-ran.

What you expected to happen

Expected Clear with DownStream + Recursive to re-run all subdag tasks.

How to reproduce

Using a slightly modified version of https://airflow.apache.org/docs/apache-airflow/stable/concepts.html#subdags:

from airflow import DAG
from airflow.example_dags.subdags.subdag import subdag
from airflow.operators.dummy import DummyOperator
from airflow.operators.subdag import SubDagOperator
from airflow.utils.dates import days_ago

def subdag(parent_dag_name, child_dag_name, args):
dag_subdag = DAG(
dag_id=f'{parent_dag_name}.{child_dag_name}',
default_args=args,
start_date=days_ago(2),
schedule_interval=None,
)

for i in range(5):
DummyOperator(
task_id='{}-task-{}'.format(child_dag_name, i + 1),
default_args=args,
dag=dag_subdag,
)

return dag_subdag

DAG_NAME = 'example_subdag_operator'

args = {
'owner': 'airflow',
}

dag = DAG(
dag_id=DAG_NAME, default_args=args, start_date=days_ago(2), schedule_interval=None, tags=['example']
)

start = DummyOperator(
task_id='start',
dag=dag,
)

section_1 = SubDagOperator(
task_id='section-1',
subdag=subdag(DAG_NAME, 'section-1', args),
dag=dag,
)

some_other_task = DummyOperator(
task_id='some-other-task',
dag=dag,
)

section_2 = SubDagOperator(
task_id='section-2',
subdag=subdag(DAG_NAME, 'section-2', args),
dag=dag,
)

end = DummyOperator(
task_id='end',
dag=dag,
)

start >> section_1 >> some_other_task >> section_2 >> end

Run the subdag fully.

Clear (with recursive/downstream) any of the SubDags.

The Subdag will be marked successful, but if you zoom into the subdag, you'll see all the child tasks were not run.

Anything else

This appears to be the same issue as the "resolved" bug #13295 . Airflow version 2.0.2 does not appear to fix the issue and neither does version 2.1.0 or 2.1.4. I am uncertain of how the fix (#14776) was shown as the key to fixing this.

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