KubernetesExecutor Airflow 2.0.0 different AirflowExceptions and tasks not getting scheduled

[lfreina]

Apache Airflow version:
``

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

Client Version: version.Info{Major:"1", Minor:"20", GitVersion:"v1.20.5", GitCommit:"6b1d87acf3c8253c123756b9e61dac642678305f", GitTreeState:"clean", BuildDate:"2021-03-31T15:33:39Z", GoVersion:"go1.15.10", Compiler:"gc", Platform:"linux/amd64"}
Server Version: version.Info{Major:"1", Minor:"19", GitVersion:"v1.19.0", GitCommit:"e19964183377d0ec2052d1f1fa930c4d7575bd50", GitTreeState:"clean", BuildDate:"2020-08-26T14:23:04Z", GoVersion:"go1.15", Compiler:"gc", Platform:"linux/amd64"}

Environment:

• Cloud provider or hardware configuration: AWS
• Others: Building my own docker image with the official documentation with Python 3.6

ARG PYTHON_VERSION=3.6
FROM python:${PYTHON_VERSION}-slim

# install deps
RUN apt-get update -y && apt-get install -y \
    libczmq-dev \
    libssl-dev \
    inetutils-telnet \
    bind9utils \
    gcc \
    build-essential \
    && apt-get clean

RUN pip install --upgrade pip

RUN pip install "apache-airflow==2.0.0" --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.0.0/constraints-3.6.txt"

RUN pip install	psycopg2-binary \
                apache-airflow-providers-cncf-kubernetes

# host specific configuration used in airflow config file
COPY airflow_local_settings.py /usr/local/lib/python3.6/airflow_local_settings.py

COPY airflow-test-env-init.sh /tmp/airflow-test-env-init.sh

COPY bootstrap.sh /bootstrap.sh
RUN chmod +x /bootstrap.sh
ENTRYPOINT ["/bootstrap.sh"]

What happened:

The tasks are not getting scheduled. In order to try the scheduler I did a simple DummyOperator DAG:

from airflow import DAG
from airflow.operators.dummy import DummyOperator


def print_hello():
    return 'Hello world!'


def print_goodbye():
    return 'Goodbye world'


dag = DAG('hello_world', description='Simple tutorial DAG',
          schedule_interval='0 12 * * *',
          start_date=datetime(2017, 3, 20), catchup=False)

dummy_operator = DummyOperator(task_id='dummy_task', retries=3, dag=dag)

dummy_operator2 = DummyOperator(task_id='dummy_task2', retries=3, dag=dag)

dummy_operator >> dummy_operator2

I tested the following getting different results:

1. airflow tasks test hello_world dummy_task 2021-01-01 --> No error at all

2. airflow tasks run hello_world dummy_task 2021-01-01

root@airflow-d78b6b585-wfl2q:/# airflow tasks run hello_world dummy_task 2021-01-01
[2021-04-08 09:15:44,383] {dagbag.py:440} INFO - Filling up the DagBag from /root/airflow/dags
Running <TaskInstance: hello_world.dummy_task 2021-01-01T00:00:00+00:00 [success]> on host 192.168.4.96
Traceback (most recent call last):
  File "/usr/local/bin/airflow", line 8, in <module>
    sys.exit(main())
  File "/usr/local/lib/python3.6/site-packages/airflow/__main__.py", line 40, in main
    args.func(args)
  File "/usr/local/lib/python3.6/site-packages/airflow/cli/cli_parser.py", line 48, in command
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/airflow/utils/cli.py", line 89, in wrapper
    return f(*args, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/airflow/cli/commands/task_command.py", line 234, in task_run
    _run_task_by_selected_method(args, dag, ti)
  File "/usr/local/lib/python3.6/site-packages/airflow/cli/commands/task_command.py", line 68, in _run_task_by_selected_method
    _run_task_by_executor(args, dag, ti)
  File "/usr/local/lib/python3.6/site-packages/airflow/cli/commands/task_command.py", line 91, in _run_task_by_executor
    executor.start()
  File "/usr/local/lib/python3.6/site-packages/airflow/executors/kubernetes_executor.py", line 493, in start
    raise AirflowException("Could not get scheduler_job_id")
airflow.exceptions.AirflowException: Could not get scheduler_job_id

3. airflow tasks run --raw hello_world dummy_task 2021-01-01

root@airflow-d78b6b585-wfl2q:/# airflow tasks run --raw hello_world dummy_task 2021-01-01
[2021-04-08 09:17:52,261] {dagbag.py:440} INFO - Filling up the DagBag from /root/airflow/dags
Running <TaskInstance: hello_world.dummy_task 2021-01-01T00:00:00+00:00 [success]> on host 192.168.4.96
Traceback (most recent call last):
  File "/usr/local/bin/airflow", line 8, in <module>
    sys.exit(main())
  File "/usr/local/lib/python3.6/site-packages/airflow/__main__.py", line 40, in main
    args.func(args)
  File "/usr/local/lib/python3.6/site-packages/airflow/cli/cli_parser.py", line 48, in command
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/airflow/utils/cli.py", line 89, in wrapper
    return f(*args, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/airflow/cli/commands/task_command.py", line 234, in task_run
    _run_task_by_selected_method(args, dag, ti)
  File "/usr/local/lib/python3.6/site-packages/airflow/cli/commands/task_command.py", line 66, in _run_task_by_selected_method
    _run_raw_task(args, ti)
  File "/usr/local/lib/python3.6/site-packages/airflow/cli/commands/task_command.py", line 151, in _run_raw_task
    pool=args.pool,
  File "/usr/local/lib/python3.6/site-packages/airflow/utils/session.py", line 65, in wrapper
    return func(*args, session=session, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/airflow/models/taskinstance.py", line 1152, in _run_raw_task
    self._run_mini_scheduler_on_child_tasks(session)
  File "/usr/local/lib/python3.6/site-packages/airflow/utils/session.py", line 62, in wrapper
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/airflow/models/taskinstance.py", line 1165, in _run_mini_scheduler_on_child_tasks
    execution_date=self.execution_date,
  File "/usr/local/lib/python3.6/site-packages/sqlalchemy/orm/query.py", line 3500, in one
    raise orm_exc.NoResultFound("No row was found for one()")
sqlalchemy.orm.exc.NoResultFound: No row was found for one()

What you expected to happen:

The tasks should run completely. Just the first one is marked as successful and the next task is never added to the scheduler.

No idea what went wrong. I already patch the code with: https://github.com/apache/airflow/pull/14160/files which I found from another similar issue.

How to reproduce it:

Use minikube and use the following setup:

airflow.yaml.txt

configmaps.yaml.txt

postgres.yaml.txt

secrets.yaml.txt

volumes.yaml.txt

namespace.yaml.txt

deploy.sh.txt

run_this_one.sh.txt

I already provided the DAG and the image file before.

Anything else we need to know:
It occurs every time with the same output even when recreated with minikube.

[boring-cyborg]

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

[kaxil]

Fixed by #14160 and released in Airflow 2.0.2 (https://pypi.org/project/apache-airflow/2.0.2/)

[lfreina]

I have now updated my instance to Airflow 2.0.2 and this bug is now gone. A new one is present though. Thanks for the fix! 👍