Task of dag stuck in queued state forever when using kubernetes executor, and triggered with a future execution date

[madcolonel10]

Airflow version: 1.10.12

import random
import string

import requests
from airflow import DAG
from airflow.contrib.sensors.python_sensor import PythonSensor
from airflow.operators.dummy_operator import DummyOperator
from datetime import datetime

dag = DAG(
    dag_id='newDebugDag',
    description='A simple tutorial DAG',
    schedule_interval=None,
    start_date=datetime(2020, 10, 30, 0, 0)
)

start = DummyOperator(task_id='run_this_first', dag=dag)


def wait_for_approval(dag_run, **context):
    software_id = dag_run.conf.get("softwareId", "Some Software Id")
    print("polling status for softwareId:{}".format(software_id))
    response = requests.get('https://5f8582eec29abd0016190be2.mockapi.io/api/v1/status')
    print(response.json())
    context["task_instance"].xcom_push(key="approverId", value="visardan-" + random.choice(string.ascii_letters))
    return response.json()[0]['status']


wait_for_approval = PythonSensor(
    task_id="wait_for_approval_debug",
    python_callable=wait_for_approval,
    dag=dag,
    poke_interval=30,
    provide_context=True,
    executor_config={"KubernetesExecutor": {"image": "apache/airflow:1.10.12-python3.6"}}
)

start >> wait_for_approval

{
    "run_id":"run-eventId109bnfghak2",
    "conf":{
        "softwareId":"something"
    },
    "execution_date": "2020-11-04 10:38:43" //any future execution date
}

The first task executes fine (dummy start task) but the dag is stuck at second task. I can see the worker pod spun up and getting completed but the task status never gets updated and ui still says queued and dag is always in running state due to this.

[boring-cyborg]

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

[pageldev]

duplicate of #13542 ? Are you still seeing this in Airflow 2.0 ?

[JieChenAtPonyai]

I met a similar issue. After investigation, I find the root cause is the the worker pod does not update the status of task to RUNNING because this dependence RunnableExecDateDep is not met. So this task is stuck in queued state.
I think enabling [allow_trigger_in_future]（

airflow/airflow/config_templates/default_airflow.cfg

Line 928 in f2558cd

allow_trigger_in_future = False

) can resolve it.