Scheduler crashes when unpausing some dags with: TypeError: '>' not supported between instances of 'NoneType' and 'int'

[lgwacker]

Apache Airflow version:
2.0.0

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

• Cloud provider or hardware configuration:
  GKE
• OS (e.g. from /etc/os-release):
  Ubuntu 18.04

What happened:
I just migrated from 1.10.14 to 2.0.0. When I turn on some random dags, the scheduler crashes with the following error:

Traceback (most recent call last):
  File "/usr/local/lib/python3.6/dist-packages/airflow/jobs/scheduler_job.py", line 1275, in _execute
    self._run_scheduler_loop()
  File "/usr/local/lib/python3.6/dist-packages/airflow/jobs/scheduler_job.py", line 1377, in _run_scheduler_loop
    num_queued_tis = self._do_scheduling(session)
  File "/usr/local/lib/python3.6/dist-packages/airflow/jobs/scheduler_job.py", line 1533, in _do_scheduling
    num_queued_tis = self._critical_section_execute_task_instances(session=session)
  File "/usr/local/lib/python3.6/dist-packages/airflow/jobs/scheduler_job.py", line 1132, in _critical_section_execute_task_instances
    queued_tis = self._executable_task_instances_to_queued(max_tis, session=session)
  File "/usr/local/lib/python3.6/dist-packages/airflow/utils/session.py", line 62, in wrapper
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.6/dist-packages/airflow/jobs/scheduler_job.py", line 1034, in _executable_task_instances_to_queued
    if task_instance.pool_slots > open_slots:
TypeError: '>' not supported between instances of 'NoneType' and 'int'

What you expected to happen:

I expected those dags would have their tasks scheduled without problems.

How to reproduce it:

Can't reproduce it yet. Still trying to figure out if this happens only with specific dags or not.

Anything else we need to know:

I couldn't find in which context task_instance.pool_slots could be None

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template!

[kaxil]

Please let us know if this is reproducible

[scrawfor]

Experiencing this same issue. We upgraded lower environments without problems, but production is throwing this error.

[kaxil]

Experiencing this same issue. We upgraded lower environments without problems, but production is throwing this error.

Can you provide some logs please with the stacktrace-- (including logs before and after the error) @scrawfor

[lgwacker]

I forgot to post an update here. I think I found out what was causing this. If you had a DAG with tasks on running state before the upgrade, when you turn those on after upgrading, this error happens.

Try clearing the whole dag run before unpausing.

[scrawfor]

@kaxil Sure.
@lgwacker - Unfortunately that did not work for me. Even renaming the dag doesn't seem to have solved the issue. Renaming the dag did fix my issue, although I had to restart the service twice.

  ____________       _____________
 ____    |__( )_________  __/__  /________      __
____  /| |_  /__  ___/_  /_ __  /_  __ \_ | /| / /
___  ___ |  / _  /   _  __/ _  / / /_/ /_ |/ |/ /
 _/_/  |_/_/  /_/    /_/    /_/  \____/____/|__/
[2021-02-02 15:03:30,310] {scheduler_job.py:1241} INFO - Starting the scheduler
[2021-02-02 15:03:30,310] {scheduler_job.py:1246} INFO - Processing each file at most -1 times
[2021-02-02 15:03:30,413] {dag_processing.py:250} INFO - Launched DagFileProcessorManager with pid: 118
[2021-02-02 15:03:30,414] {scheduler_job.py:1751} INFO - Resetting orphaned tasks for active dag runs
[2021-02-02 15:03:30,463] {settings.py:52} INFO - Configured default timezone Timezone('America/New_York')
[2021-02-02 15:03:30,643] {scheduler_job.py:938} INFO - 32 tasks up for execution:
       <TASK LIST WAS HERE>
[2021-02-02 15:03:30,652] {scheduler_job.py:967} INFO - Figuring out tasks to run in Pool(name=default_pool) with 128 open slots and 32 task instances ready to be queued
[2021-02-02 15:03:30,652] {scheduler_job.py:995} INFO - DAG <dag1> has 0/16 running and queued tasks
[2021-02-02 15:03:30,652] {scheduler_job.py:995} INFO - DAG <dag1> has 1/16 running and queued tasks
[2021-02-02 15:03:30,652] {scheduler_job.py:995} INFO - DAG <dag1> has 2/16 running and queued tasks
[2021-02-02 15:03:30,653] {scheduler_job.py:995} INFO - DAG <dag1> has 3/16 running and queued tasks
[2021-02-02 15:03:30,653] {scheduler_job.py:995} INFO - DAG <dag1> has 4/16 running and queued tasks
[2021-02-02 15:03:30,653] {scheduler_job.py:995} INFO - DAG <dag1> has 5/16 running and queued tasks
[2021-02-02 15:03:30,653] {scheduler_job.py:995} INFO - DAG <dag2> has 0/16 running and queued tasks
[2021-02-02 15:03:30,653] {scheduler_job.py:995} INFO - DAG <dag2> has 1/16 running and queued tasks
[2021-02-02 15:03:30,654] {scheduler_job.py:995} INFO - DAG <dag2> has 2/16 running and queued tasks
[2021-02-02 15:03:30,654] {scheduler_job.py:995} INFO - DAG <dag2> has 3/16 running and queued tasks
[2021-02-02 15:03:30,654] {scheduler_job.py:995} INFO - DAG <dag2> has 4/16 running and queued tasks
[2021-02-02 15:03:30,661] {scheduler_job.py:1293} ERROR - Exception when executing SchedulerJob._run_scheduler_loop
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1275, in _execute
    self._run_scheduler_loop()
  File "/home/airflow/.local/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/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1533, in _do_scheduling
    num_queued_tis = self._critical_section_execute_task_instances(session=session)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1132, in _critical_section_execute_task_instances
    queued_tis = self._executable_task_instances_to_queued(max_tis, session=session)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/session.py", line 62, in wrapper
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1034, in _executable_task_instances_to_queued
    if task_instance.pool_slots > open_slots:
TypeError: '>' not supported between instances of 'NoneType' and 'int'
[2021-02-02 15:03:31,668] {process_utils.py:95} INFO - Sending Signals.SIGTERM to GPID 118
[2021-02-02 15:03:32,157] {process_utils.py:61} INFO - Process psutil.Process(pid=164, status='terminated', started='15:03:30') (164) terminated with exit code None
[2021-02-02 15:03:32,175] {process_utils.py:61} INFO - Process psutil.Process(pid=172, status='terminated', started='15:03:30') (172) terminated with exit code None
[2021-02-02 15:03:32,177] {process_utils.py:201} INFO - Waiting up to 5 seconds for processes to exit...
[2021-02-02 15:03:32,188] {process_utils.py:61} INFO - Process psutil.Process(pid=118, status='terminated', exitcode=0, started='15:03:29') (118) terminated with exit code 0
[2021-02-02 15:03:32,189] {scheduler_job.py:1296} INFO - Exited execute loop
Process QueuedLocalWorker-29:
Process QueuedLocalWorker-31:
Process QueuedLocalWorker-33:
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/local_executor.py", line 69, in run
    return super().run()
  File "/usr/local/lib/python3.8/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/local_executor.py", line 176, in do_work
    key, command = self.task_queue.get()
  File "<string>", line 2, in get
  File "/usr/local/lib/python3.8/multiprocessing/managers.py", line 835, in _callmethod
    kind, result = conn.recv()
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 250, in recv
    buf = self._recv_bytes()
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 414, in _recv_bytes
    buf = self._recv(4)
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 383, in _recv
    raise EOFError
EOFError
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/local_executor.py", line 69, in run
    return super().run()
  File "/usr/local/lib/python3.8/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/local_executor.py", line 176, in do_work
    key, command = self.task_queue.get()
  File "<string>", line 2, in get
  File "/usr/local/lib/python3.8/multiprocessing/managers.py", line 835, in _callmethod
    kind, result = conn.recv()
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 250, in recv
    buf = self._recv_bytes()
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 414, in _recv_bytes
    buf = self._recv(4)
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 383, in _recv
    raise EOFError
EOFError
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/local_executor.py", line 69, in run
    return super().run()
  File "/usr/local/lib/python3.8/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/local_executor.py", line 176, in do_work
    key, command = self.task_queue.get()
  File "<string>", line 2, in get
  File "/usr/local/lib/python3.8/multiprocessing/managers.py", line 835, in _callmethod
    kind, result = conn.recv()
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 250, in recv
    buf = self._recv_bytes()
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 414, in _recv_bytes
    buf = self._recv(4)
  File "/usr/local/lib/python3.8/multiprocessing/connection.py", line 383, in _recv
    raise EOFError
EOFError
Process QueuedLocalWorker-32:
Process QueuedLocalWorker-26:
Process QueuedLocalWorker-30:

Env Details:

• Docker Image: apache/airflow:2.0.0-python3.8
• Postgres Metadata DB
• Local Executor

[lgwacker]

@scrawfor another thing I did was deleting the pool, restarting the scheduler, turning the dag on, and creating the pool again. Can you try that?

Besides that, do you have any DAGs that were previously deleted still showing up on the webserver (since now it serializes them)? If so, delete them. This caused problems for me too.

[scrawfor]

@lgwacker In a rush to get back online I just renamed the dag which did work after a couple of restarts. But I appreciate the help!

[kaxil]

Another workaround is to run the following query in your metadata db:

UPDATE task_instance SET pool_slots = 1 WHERE pool_slots IS NULL;

[gdevanla]

Thanks, @kaxil . That update statement has resolved the problem for me.

[ashb]

@kaxil Do we have any idea how this might have possibly happened? Bad/race-y previous migration?