<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.1.2</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Official docker image</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Official Apache Airflow Helm Chart</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">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"</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">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.</p>
<p dir="auto">Cleanup Done:</p>
<p dir="auto">UPDATE public.dag_run SET conf= NULL;<br>
DELETE FROM public.dag_pickle;</p>
<p dir="auto">Logs:</p>
<hr>
<p dir="auto">____ |<strong>( )</strong>_______ <strong>/</strong> /________ __<br>
____ /| |_ /__ <em><em><em>/</em> /</em> __ /</em> __ _ | /| / /<br>
___ ___ | / _ / _ <strong>/ _ / / /<em>/ /</em> |/ |/ /<br>
<em>/</em>/ |<em>/</em>/ /<em>/ /</em>/ /<em>/ _</em></strong>/____/|__/<br>
[2021-08-31 16:20:29,139] {scheduler_job.py:1266} INFO - Starting the scheduler<br>
[2021-08-31 16:20:29,139] {scheduler_job.py:1271} INFO - Processing each file at most -1 times<br>
[2021-08-31 16:20:29,523] {dag_processing.py:254} INFO - Launched DagFileProcessorManager with pid: 19<br>
[2021-08-31 16:20:29,528] {scheduler_job.py:1835} INFO - Resetting orphaned tasks for active dag runs<br>
[2021-08-31 16:20:29,539] {settings.py:51} INFO - Configured default timezone Timezone('UTC')<br>
[2021-08-31 16:20:29,628] {scheduler_job.py:1315} ERROR - Exception when executing SchedulerJob._run_scheduler_loop<br>
Traceback (most recent call last):<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1299, in _execute<br>
self._run_scheduler_loop()<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1392, in _run_scheduler_loop<br>
num_queued_tis = self._do_scheduling(session)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1531, in _do_scheduling<br>
self._schedule_dag_run(dag_run, active_runs_by_dag_id.get(dag_run.dag_id, set()), session)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1759, in _schedule_dag_run<br>
self._verify_integrity_if_dag_changed(dag_run=dag_run, session=session)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper<br>
return func(*args, **kwargs)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1785, in <em>verify_integrity_if_dag_changed<br>
dag_run.verify_integrity(session=session)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper<br>
return func(*args, **kwargs)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/dagrun.py", line 638, in verify_integrity<br>
tis = self.get_task_instances(session=session)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper<br>
return func(*args, **kwargs)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/dagrun.py", line 328, in get_task_instances<br>
return tis.all()<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/query.py", line 3373, in all<br>
return list(self)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 100, in instances<br>
cursor.close()<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/util/langhelpers.py", line 70, in <strong>exit</strong><br>
with_traceback=exc_tb,<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 182, in raise</em><br>
raise exception<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 80, in instances<br>
rows = [proc(row) for row in fetch]<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 80, in <br>
rows = [proc(row) for row in fetch]<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 588, in _instance<br>
populators,<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 725, in <em>populate_full<br>
dict</em>[key] = getter(row)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/sql/sqltypes.py", line 1723, in process<br>
return loads(value)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 275, in loads<br>
return load(file, ignore, **kwds)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 270, in load<br>
return Unpickler(file, ignore=ignore, **kwds).load()<br>
File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 472, in load<br>
obj = StockUnpickler.load(self)<br>
ValueError: unsupported pickle protocol: 5<br>
[2021-08-31 16:20:30,634] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 19<br>
[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<br>
[2021-08-31 16:20:30,970] {scheduler_job.py:1326} INFO - Exited execute loop<br>
Traceback (most recent call last):<br>
File "/home/airflow/.local/bin/airflow", line 8, in <br>
sys.exit(main())<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/<strong>main</strong>.py", line 40, in main<br>
args.func(args)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/cli/cli_parser.py", line 48, in command<br>
return func(*args, **kwargs)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/cli.py", line 91, in wrapper<br>
return f(*args, **kwargs)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/cli/commands/scheduler_command.py", line 64, in scheduler<br>
job.run()<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/base_job.py", line 245, in run<br>
self._execute()<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1299, in _execute<br>
self._run_scheduler_loop()<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1392, in _run_scheduler_loop<br>
num_queued_tis = self._do_scheduling(session)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1531, in _do_scheduling<br>
self._schedule_dag_run(dag_run, active_runs_by_dag_id.get(dag_run.dag_id, set()), session)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1759, in _schedule_dag_run<br>
self._verify_integrity_if_dag_changed(dag_run=dag_run, session=session)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper<br>
return func(*args, **kwargs)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1785, in <em>verify_integrity_if_dag_changed<br>
dag_run.verify_integrity(session=session)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper<br>
return func(*args, **kwargs)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/dagrun.py", line 638, in verify_integrity<br>
tis = self.get_task_instances(session=session)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py", line 67, in wrapper<br>
return func(*args, **kwargs)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/dagrun.py", line 328, in get_task_instances<br>
return tis.all()<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/query.py", line 3373, in all<br>
return list(self)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 100, in instances<br>
cursor.close()<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/util/langhelpers.py", line 70, in <strong>exit</strong><br>
with_traceback=exc_tb,<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 182, in raise</em><br>
raise exception<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 80, in instances<br>
rows = [proc(row) for row in fetch]<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 80, in <br>
rows = [proc(row) for row in fetch]<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 588, in _instance<br>
populators,<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py", line 725, in <em>populate_full<br>
dict</em>[key] = getter(row)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/sqlalchemy/sql/sqltypes.py", line 1723, in process<br>
return loads(value)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 275, in loads<br>
return load(file, ignore, **kwds)<br>
File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 270, in load<br>
return Unpickler(file, ignore=ignore, **kwds).load()<br>
File "/home/airflow/.local/lib/python3.6/site-packages/dill/_dill.py", line 472, in load<br>
obj = StockUnpickler.load(self)<br>
ValueError: unsupported pickle protocol: 5</p>
<h3 dir="auto">What you expected to happen</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>
<p dir="auto"><strong>Apache Airflow version</strong>: 1.7.1.2</p>
<p dir="auto">Ticket was created 11/Jul/16 16:06</p>
<p dir="auto"><strong>Kubernetes version (if you are using kubernetes)</strong> (use <code class="notranslate">kubectl version</code>):</p>
<p dir="auto"><strong>Environment</strong>:</p>
<ul dir="auto">
<li><strong>Cloud provider or hardware configuration</strong>:</li>
<li><strong>OS</strong> (e.g. from /etc/os-release):</li>
<li><strong>Kernel</strong> (e.g. <code class="notranslate">uname -a</code>):</li>
<li><strong>Install tools</strong>:</li>
<li><strong>Others</strong>:<br>
<strong>What happened</strong>:</li>
</ul>
<p dir="auto">Currently, there is an issue of tasks overlapping between DAG runs<br>
This can cause tasks to compete for resources as well as duplicating or overwriting what the other task is doing.</p>
<p dir="auto"><strong>What you expected to happen</strong>:</p>
<p dir="auto">As a the Airflow administrator,<br>
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.</p>
<p dir="auto"><strong>How to reproduce it</strong>:</p>
<ol dir="auto">
<li>Create a DAG with a short scheduled interval</li>
<li>Create a task in that DAG to run longer than the interval<br>
Result: Both tasks end up running that the same time.</li>
</ol>
<p dir="auto"><strong>Anything else we need to know</strong>:</p>
<p dir="auto">Moved here from <a href="https://issues.apache.org/jira/browse/AIRFLOW-323" rel="nofollow">https://issues.apache.org/jira/browse/AIRFLOW-323</a></p>