Issue 1:
Apache Airflow version

2.1.2

Operating System

Official docker image

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

Deployment details

Originally had airflow 2.0.2 installed via 3rd party helm chart (running external postgres db). I uninstalled it and installed airflow 2.1.2 using the official chart. Pods are alll started up except the schedular pod is crashing with "ValueError: unsupported pickle protocol: 5"

What happened

I get a "ValueError: unsupported pickle protocol: 5". I went into the database and did some cleanup but still getting error. Not sure if there are other tables/colums that also need to be cleaned up.

Cleanup Done:

UPDATE public.dag_run SET conf= NULL;

DELETE FROM public.dag_pickle;

Logs:

____ |( )_______ / /________ __

____ /| |_ /__ / / __ / __ _ | /| / /

___ ___ | / _ / _ / _ / / // / |/ |/ /

// |// // // // _/____/|__/

[2021-08-31 16:20:29,139] {scheduler_job.py:1266} INFO - Starting the scheduler

[2021-08-31 16:20:29,139] {scheduler_job.py:1271} INFO - Processing each file at most -1 times

[2021-08-31 16:20:29,523] {dag_processing.py:254} INFO - Launched DagFileProcessorManager with pid: 19

[2021-08-31 16:20:29,528] {scheduler_job.py:1835} INFO - Resetting orphaned tasks for active dag runs

[2021-08-31 16:20:29,539] {settings.py:51} INFO - Configured default timezone Timezone('UTC')

[2021-08-31 16:20:29,628] {scheduler_job.py:1315} ERROR - Exception when executing SchedulerJob._run_scheduler_loop

Traceback (most recent call last):

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1299, in _execute

self._run_scheduler_loop()

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1392, in _run_scheduler_loop

num_queued_tis = self._do_scheduling(session)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1531, in _do_scheduling

self._schedule_dag_run(dag_run, active_runs_by_dag_id.get(dag_run.dag_id, set()), session)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1759, in _schedule_dag_run

self._verify_integrity_if_dag_changed(dag_run=dag_run, session=session)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper

return func(*args, **kwargs)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1785, in verify_integrity_if_dag_changed

dag_run.verify_integrity(session=session)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper

return func(*args, **kwargs)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/dagrun.py", line 638, in verify_integrity

tis = self.get_task_instances(session=session)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper

return func(*args, **kwargs)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/dagrun.py", line 328, in get_task_instances

return tis.all()

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/query.py", line 3373, in all

return list(self)

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 100, in instances

cursor.close()

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/util/langhelpers.py", line 70, in exit

with_traceback=exc_tb,

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 182, in raise

raise exception

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 80, in instances

rows = [proc(row) for row in fetch]

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 80, in

rows = [proc(row) for row in fetch]

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 588, in _instance

populators,

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 725, in populate_full

dict[key] = getter(row)

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/sql/sqltypes.py", line 1723, in process

return loads(value)

File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 275, in loads

return load(file, ignore, **kwds)

File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 270, in load

return Unpickler(file, ignore=ignore, **kwds).load()

File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 472, in load

obj = StockUnpickler.load(self)

ValueError: unsupported pickle protocol: 5

[2021-08-31 16:20:30,634] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 19

[2021-08-31 16:20:30,970] {process_utils.py:66} INFO - Process psutil.Process(pid=19, status='terminated', exitcode=0, started='16:20:28') (19) terminated with exit code 0

[2021-08-31 16:20:30,970] {scheduler_job.py:1326} INFO - Exited execute loop

Traceback (most recent call last):

File "/home/airflow/.local/bin/airflow", line 8, in

sys.exit(main())

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/main.py", line 40, in main

args.func(args)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/cli/cli_parser.py", line 48, in command

return func(*args, **kwargs)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/cli.py", line 91, in wrapper

return f(*args, **kwargs)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/cli/commands/scheduler_command.py", line 64, in scheduler

job.run()

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/base_job.py", line 245, in run

self._execute()

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1299, in _execute

self._run_scheduler_loop()

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1392, in _run_scheduler_loop

num_queued_tis = self._do_scheduling(session)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1531, in _do_scheduling

self._schedule_dag_run(dag_run, active_runs_by_dag_id.get(dag_run.dag_id, set()), session)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1759, in _schedule_dag_run

self._verify_integrity_if_dag_changed(dag_run=dag_run, session=session)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper

return func(*args, **kwargs)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1785, in verify_integrity_if_dag_changed

dag_run.verify_integrity(session=session)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper

return func(*args, **kwargs)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/dagrun.py", line 638, in verify_integrity

tis = self.get_task_instances(session=session)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper

return func(*args, **kwargs)

File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/dagrun.py", line 328, in get_task_instances

return tis.all()

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/query.py", line 3373, in all

return list(self)

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 100, in instances

cursor.close()

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/util/langhelpers.py", line 70, in exit

with_traceback=exc_tb,

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 182, in raise

raise exception

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 80, in instances

rows = [proc(row) for row in fetch]

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 80, in

rows = [proc(row) for row in fetch]

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 588, in _instance

populators,

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 725, in populate_full

dict[key] = getter(row)

File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/sql/sqltypes.py", line 1723, in process

return loads(value)

File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 275, in loads

return load(file, ignore, **kwds)

File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 270, in load

return Unpickler(file, ignore=ignore, **kwds).load()

File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 472, in load

obj = StockUnpickler.load(self)

ValueError: unsupported pickle protocol: 5

What you expected to happen

No response

How to reproduce

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: 1.7.1.2

Ticket was created 11/Jul/16 16:06

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

Environment:

Cloud provider or hardware configuration:

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

Kernel (e.g. uname -a):

Install tools:

Others:

What happened:

Currently, there is an issue of tasks overlapping between DAG runs

This can cause tasks to compete for resources as well as duplicating or overwriting what the other task is doing.

What you expected to happen:

As a the Airflow administrator,

If a task from a previous DAG Run is still running when the next scheduled run triggers the same task, there should be a way prevent the tasks from overlapping.

How to reproduce it:

Create a DAG with a short scheduled interval

Create a task in that DAG to run longer than the interval

Result: Both tasks end up running that the same time.

Anything else we need to know:

Moved here from https://issues.apache.org/jira/browse/AIRFLOW-323