Issue 1:
Apache Airflow version

Other Airflow 2 version (please specify below)

What happened

AirFlow version 2.3.3.

When we use TaskFlow API for the DAG and have some task with trigger_rule="none_failed" this task fails when its predecessor is skipped. The error occurs during XCom read:

File "/usr/local/lib/python3.8/site-packages/airflow/models/xcom_arg.py", line 152, in resolve
raise AirflowException(
airflow.exceptions.AirflowException: XComArg result from exec at skip_test with key="return_value" is not found!

What you think should happen instead

It should execute its code with empty input. This behaviour existed in version 2.2.3. Also it's well aligned with behaviour of general tasks connected by arrows: downstream task executes successfully with empty input.

How to reproduce

Create a DAG.

Add @task-decorated task that raises AirflowSkipException

Add another @task-decorated task that checks if its input is empty.

Pass an output of the first task to the second task.

Add DummyOperator task as a downstream of the first task. trigger_rule="none_failed"

Execute a DAG.

The second task fails.

from airflow.decorators import dag, task
from airflow.utils.dates import days_ago
from airflow.exceptions import AirflowSkipException
from airflow.operators.dummy import DummyOperator

@dag(start_date=days_ago(1))
def skip_test():

@task
def exec():
raise AirflowSkipException("skipping")

# This will fail
@task(trigger_rule="none_failed")
def finish(prev_res):
print(prev_res is None)

# While this will not
finish_2 = DummyOperator(
task_id = "finish_2",
trigger_rule="none_failed"
)

exec_res = exec()
finish(exec_res)
exec_res >> finish_2

return

G_DUMMY = skip_test()

Operating System

Debian GNU/Linux 10 (buster)

Versions of Apache Airflow Providers

No response

Deployment

Docker-Compose

Deployment details

No response

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

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

Environment:

Cloud provider or hardware configuration: tencent cloud

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

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

Install tools:

Others: Server version: 8.0.22 MySQL Community Server - GPL

What happened:

Scheduler died when I try to modify a dag's schedule_interval from "None" to "* */1 * * *"(I edited the dag file in the dag folder and saved it). A few minutes later I tried to start the scheduler again and it began to run.

And the logs are as follows:

{2021-01-15 09:06:22,636} {{scheduler_job.py:1293}} ERROR - Exception when executing SchedulerJob._run_scheduler_loop
Traceback (most recent call last):
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1276, in _execute_context
self.dialect.do_execute(
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 609, in do_execute
cursor.execute(statement, parameters)
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/MySQLdb/cursors.py", line 209, in execute
res = self._query(query)
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/MySQLdb/cursors.py", line 315, in _query
db.query(q)
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/MySQLdb/connections.py", line 239, in query
_mysql.connection.query(self, query)
MySQLdb._exceptions.IntegrityError: (1062, "Duplicate entry 'huge_demo13499411352-2021-01-15 01:04:00.000000' for key 'dag_run.dag_id'")

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1275, in _execute
self._run_scheduler_loop()
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1377, in _run_scheduler_loop
num_queued_tis = self._do_scheduling(session)
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1474, in _do_scheduling
self._create_dag_runs(query.all(), session)
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1561, in _create_dag_runs
dag.create_dagrun(
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/utils/session.py", line 62, in wrapper
return func(*args, **kwargs)
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/models/dag.py", line 1807, in create_dagrun
session.flush()
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 2540, in flush
self._flush(objects)
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 2682, in _flush
transaction.rollback(_capture_exception=True)
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/util/langhelpers.py", line 68, in __exit__
compat.raise_(
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 182, in raise_
raise exception
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 2642, in _flush
flush_context.execute()
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 422, in execute
rec.execute(self)
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 586, in execute
persistence.save_obj(
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 239, in save_obj
_emit_insert_statements(
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 1135, in _emit_insert_statements
result = cached_connections[connection].execute(
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1011, in execute
return meth(self, multiparams, params)
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/sql/elements.py", line 298, in _execute_on_connection
return connection._execute_clauseelement(self, multiparams, params)
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1124, in _execute_clauseelement
ret = self._execute_context(
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1316, in _execute_context
self._handle_dbapi_exception(
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1510, in _handle_dbapi_exception
util.raise_(
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 182, in raise_
raise exception
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1276, in _execute_context
self.dialect.do_execute(
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 609, in do_execute
cursor.execute(statement, parameters)
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/MySQLdb/cursors.py", line 209, in execute
res = self._query(query)
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/MySQLdb/cursors.py", line 315, in _query
db.query(q)
File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/MySQLdb/connections.py", line 239, in query
_mysql.connection.query(self, query)
sqlalchemy.exc.IntegrityError: (MySQLdb._exceptions.IntegrityError) (1062, "Duplicate entry 'huge_demo13499411352-2021-01-15 01:04:00.000000' for key 'dag_run.dag_id'")
[SQL: INSERT INTO dag_run (dag_id, execution_date, start_date, end_date, state, run_id, creating_job_id, external_trigger, run_type, conf, last_scheduling_decision, dag_hash) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)]
[parameters: ('huge_demo13499411352', datetime.datetime(2021, 1, 15, 1, 4), datetime.datetime(2021, 1, 15, 1, 6, 22, 629433), None, 'running', 'scheduled__2021-01-15T01:04:00+00:00', 71466, 0, <DagRunType.SCHEDULED: 'scheduled'>, b'\x80\x05}\x94.', None, '60078c379cdeecb9bc8844eed5aa9745')]
(Background on this error at: http://sqlalche.me/e/13/gkpj)
{2021-01-15 09:06:23,648} {{process_utils.py:95}} INFO - Sending Signals.SIGTERM to GPID 66351
{2021-01-15 09:06:23,781} {{process_utils.py:61}} INFO - Process psutil.Process(pid=66351, status='terminated') (66351) terminated with exit code 0
{2021-01-15 09:06:23,781} {{scheduler_job.py:1296}} INFO - Exited execute loop

What you expected to happen:

Schdeduler should not die.

How to reproduce it:

I don't know how to reproduce it

Anything else we need to know:

No