n-badtke-cg commented on issue #71772:
URL: https://github.com/apache/airflow/issues/71772#issuecomment-5343978725
@tirkarthi Given your commands, I opened a remote shell to the Kubernetes
Pod did what you asked for:
<details>
<summary>bash commands dump (Click to expand)</summary>
```bash
$ kubectl exec -n airflow airflow-dag-processor-66894c6b74-bkp4d -c
dag-processor -it -- bash -il
airflow@airflow-dag-processor-66894c6b74-bkp4d:/opt/airflow$ cd dags
airflow@airflow-dag-processor-66894c6b74-bkp4d:/opt/airflow/dags$ cat >
gh71772.py<< EOF
> from __future__ import annotations
from datetime import datetime, timedelta
from airflow import XComArg
from airflow.sdk import dag, task
from airflow.sdk.exceptions import AirflowFailException
HEALTH_CHECK_MESSAGE: str = "Health check successful"
HEALTHY: str = "healthy"
UNHEALTHY: str = "unhealthy"
@dag(
schedule="*/2 * * * *",
start_date=datetime(2026, 7, 1),
dagrun_timeout=timedelta(minutes=3),
is_paused_upon_creation=False,
catchup=False,
)
def gh71772() -> None:
@task.bash
def health_check(text: str) -> str:
return f"echo '{text}'"
@task
def result_is_healthy(*, result: XComArg, check_against: str) -> bool:
return str(result) == check_against
@task
def print_result(is_healthy: bool) -> None:
print("is_healthy: " + str(type(is_healthy)) + " " +
str(is_healthy))
print(HEALTHY if is_healthy == True else UNHEALTHY)
if not is_healthy == True:
raise AirflowFailException()
health_check_result: XComArg = health_check(HEALTH_CHECK_MESSAGE)
result_is_healthy_bool: bool = result_is_healthy(
result=health_check_result, check_against=HEALTH_CHECK_MESSAGE
)
print_result(result_is_healthy_bool)
gh71772()
> EOF
airflow@airflow-dag-processor-66894c6b74-bkp4d:/opt/airflow/dags$ cat
gh71772.py
from __future__ import annotations
from datetime import datetime, timedelta
from airflow import XComArg
from airflow.sdk import dag, task
from airflow.sdk.exceptions import AirflowFailException
HEALTH_CHECK_MESSAGE: str = "Health check successful"
HEALTHY: str = "healthy"
UNHEALTHY: str = "unhealthy"
@dag(
schedule="*/2 * * * *",
start_date=datetime(2026, 7, 1),
dagrun_timeout=timedelta(minutes=3),
is_paused_upon_creation=False,
catchup=False,
)
def gh71772() -> None:
@task.bash
def health_check(text: str) -> str:
return f"echo '{text}'"
@task
def result_is_healthy(*, result: XComArg, check_against: str) -> bool:
return str(result) == check_against
@task
def print_result(is_healthy: bool) -> None:
print("is_healthy: " + str(type(is_healthy)) + " " +
str(is_healthy))
print(HEALTHY if is_healthy == True else UNHEALTHY)
if not is_healthy == True:
raise AirflowFailException()
health_check_result: XComArg = health_check(HEALTH_CHECK_MESSAGE)
result_is_healthy_bool: bool = result_is_healthy(
result=health_check_result, check_against=HEALTH_CHECK_MESSAGE
)
print_result(result_is_healthy_bool)
gh71772()
airflow@airflow-dag-processor-66894c6b74-bkp4d:/opt/airflow/dags$ python
Python 3.13.14 (main, Aug 12 2026, 08:57:09) [GCC 12.2.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> from airflow.dag_processing.dagbag import DagBag
...
KeyboardInterrupt
>>> from airflow.dag_processing.dagbag import DagBag
2026-08-19T14:50:49.459701Z [warning ] Both the standard OpenTelemetry
environment variables and the Airflow OpenTelemetry configs have been provided.
Using the OpenTelemetry environment variables. The Airflow configs have been
deprecated and will be removed in the future.
[airflow.sdk._shared.observability.common] loc=common.py:87
2026-08-19T14:50:49.460321Z [info ] [OTLPMetricExporter] Connecting to
OpenTelemetry Collector at
http://grafana-k8s-monitoring-receiver-collector.observability-system.svc.cluster.local:4317
[airflow.sdk._shared.observability.common] loc=common.py:134
>>> DagBag("/opt/airlfow/dags/gh71772.py")
2026-08-19T14:51:22.606820Z [info ] Filling up the DagBag from
/opt/airlfow/dags/gh71772.py [airflow.dag_processing.dagbag.DagBag]
loc=dagbag.py:460
<airflow.dag_processing.dagbag.DagBag object at 0x7ba101d701a0>
>>> DagBag("/home/karthikeyan/airflow/dags/gh71772.py").dags
2026-08-19T14:51:34.388882Z [info ] Filling up the DagBag from
/home/karthikeyan/airflow/dags/gh71772.py
[airflow.dag_processing.dagbag.DagBag] loc=dagbag.py:460
{}
>>> DagBag("/opt/airflow/dags/gh71772.py").dags
2026-08-19T14:51:47.094370Z [info ] Filling up the DagBag from
/opt/airflow/dags/gh71772.py [airflow.dag_processing.dagbag.DagBag]
loc=dagbag.py:460
{'gh71772': <DAG: gh71772>}
>>>
```
</details>
Means, the specific dag processor pods is able to parse the dag code in
itself. So why is there the timeout? Even weirder is that the dag processor
recognizes that there are python files existing, so the git fetch is also
working ??!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
