task fails post execution with "RuntimeError: This class is not fork-safe"

[maxgruber19]

Hi once again,

one of our customers has a problem with submitting dags containing two tasks after branching, we we changed probably everything but nothing really helped.

we're running airflow 2.9.3 from stackable 24.11.0 and according to @sbernauer we have the object storage backend configured for xcom and logs (maybe thats important to know). we tried with celery executors and now with k8s executors, no difference. we tried gevent, threads, prefork and solo as AIRFLOW__CELERY__POOL. We also tried setting AIRFLOW__CORE__MP_START_METHOD to spawn

the error we get is following

[2025-06-20, 11:48:29 UTC] {taskinstance.py:441} ▼ Post task execution logs
[2025-06-20, 11:48:29 UTC] {taskinstance.py:2905} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/stackable/app/lib64/python3.9/site-packages/airflow/models/taskinstance.py", line 465, in _execute_task
    result = _execute_callable(context=context, **execute_callable_kwargs)
  File "/stackable/app/lib64/python3.9/site-packages/airflow/models/taskinstance.py", line 432, in _execute_callable
    return execute_callable(context=context, **execute_callable_kwargs)
  File "/stackable/app/lib64/python3.9/site-packages/airflow/models/baseoperator.py", line 401, in wrapper
    return func(self, *args, **kwargs)
  File "/stackable/app/lib64/python3.9/site-packages/airflow/providers/cncf/kubernetes/operators/pod.py", line 591, in execute
    return self.execute_sync(context)
  File "/stackable/app/lib64/python3.9/site-packages/airflow/providers/cncf/kubernetes/operators/pod.py", line 607, in execute_sync
    ti.xcom_push(key="pod_name", value=self.pod.metadata.name)
  File "/stackable/app/lib64/python3.9/site-packages/airflow/utils/session.py", line 79, in wrapper
    return func(*args, session=session, **kwargs)
  File "/stackable/app/lib64/python3.9/site-packages/airflow/models/taskinstance.py", line 3197, in xcom_push
    XCom.set(
  File "/stackable/app/lib64/python3.9/site-packages/airflow/utils/session.py", line 76, in wrapper
    return func(*args, **kwargs)
  File "/stackable/app/lib64/python3.9/site-packages/airflow/models/xcom.py", line 246, in set
    value = cls.serialize_value(
  File "/stackable/app/lib64/python3.9/site-packages/airflow/providers/common/io/xcom/backend.py", line 142, in serialize_value
    if not p.exists():
  File "/stackable/app/lib64/python3.9/site-packages/upath/core.py", line 711, in exists
    return self.fs.exists(self.path)
  File "/stackable/app/lib64/python3.9/site-packages/fsspec/asyn.py", line 118, in wrapper
    return sync(self.loop, func, *args, **kwargs)
  File "/stackable/app/lib64/python3.9/site-packages/fsspec/asyn.py", line 328, in loop
    raise RuntimeError("This class is not fork-safe")
RuntimeError: This class is not fork-safe

dag code has to be "blurred" for typical reasons but it looks like that, kpo = kubernetespodoperator

(
    kpo_a
    >> kpo_b
    >> kpo_c
    >> kpo_d
    >> branch_x
)

branch_x >> [kpo_e, kpo_f] 
kpo_e >> kpo_h

in this dag the kpo_f is the one thats always failing although it does not have any configuration differences, the pod is spawning and working but airflow seems to have a problem with it somehow.

please let me know what else you need to dig deeper, its a very ugly problem i know

[adwk67]

Hi @maxgruber19, as expected, this is proving difficult to reproduce.
This is just about the simplest DAG I can come up to replicate the basic task structure (based off one of the example DAGs), but it works without problem.

    from __future__ import annotations

    import pendulum
    from airflow.providers.cncf.kubernetes.operators.pod import (
        KubernetesPodOperator,
    )
    from airflow.sdk import DAG, task


    @task.branch()
    def should_run(**kwargs) -> str:
        """
        Determine which empty_task should be run based on if the logical date minute is even or odd.

        :param dict kwargs: Context
        :return: Id of the task to run
        """
        print(f"------------- exec dttm = {kwargs['logical_date']} and minute = {kwargs['logical_date'].minute}")
        if kwargs["logical_date"].minute % 2 == 0:
            return "branch_task_3"
        return "branch_task_4"


    with DAG(
        dag_id="branching_test",
        schedule="*/1 * * * *",
        start_date=pendulum.datetime(2025, 1, 1, tz="UTC"),
        catchup=False,
        default_args={"depends_on_past": True},
        tags=["example"],
    ) as dag:
        cond = should_run()

        task_1 = KubernetesPodOperator(
                          name=f"branch_task_1",
                          image="oci.stackable.tech/sdp/testing-tools:0.2.0-stackable0.0.0-dev",
                          cmds=["python"],
                          arguments=["-V"],
                          image_pull_policy="IfNotPresent",
                          task_id="branch_task_1",
                          do_xcom_push=False
                      )
        task_2 = KubernetesPodOperator(
                          name=f"branch_task_2",
                          image="oci.stackable.tech/sdp/testing-tools:0.2.0-stackable0.0.0-dev",
                          cmds=["python"],
                          arguments=["-V"],
                          image_pull_policy="IfNotPresent",
                          task_id="branch_task_2",
                          do_xcom_push=False
                      )
        task_3 = KubernetesPodOperator(
                          name=f"branch_task_3",
                          image="oci.stackable.tech/sdp/testing-tools:0.2.0-stackable0.0.0-dev",
                          cmds=["python"],
                          arguments=["-V"],
                          image_pull_policy="IfNotPresent",
                          task_id="branch_task_3",
                          do_xcom_push=False
                      )
        task_4 = KubernetesPodOperator(
                          name=f"branch_task_4",
                          image="oci.stackable.tech/sdp/testing-tools:0.2.0-stackable0.0.0-dev",
                          cmds=["python"],
                          arguments=["-V"],
                          image_pull_policy="IfNotPresent",
                          task_id="branch_task_4",
                          do_xcom_push=False
                      )
        task_5 = KubernetesPodOperator(
                          name=f"branch_task_5",
                          image="oci.stackable.tech/sdp/testing-tools:0.2.0-stackable0.0.0-dev",
                          cmds=["python"],
                          arguments=["-V"],
                          image_pull_policy="IfNotPresent",
                          task_id="branch_task_5",
                          do_xcom_push=False
                      )
        
        task_1 >> task_2 >> cond
        cond >> [task_3, task_4]
        task_3 >> task_5

Do you have access to the target environment? If so, can you deactivate the xcom pushes (or are they required by the tasks themselves)?

[adwk67]

I added in the xcom push steps and added an extra permission. Tested with 24.11 and Airflow 2.10.2:

---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: airflow-xcom-clusterrole
rules:
- apiGroups:
  - ""
  resources:
  - pods/exec
  verbs:
  - create
  - get
  - list
  - watch
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: airflow-xcom-clusterrole-binding
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: airflow-xcom-clusterrole
subjects:
- apiGroup: rbac.authorization.k8s.io
  kind: Group
  name: system:serviceaccounts

[adwk67]

Can you tell me how/where your xcom data is stored? (the only information I can find related to the error message seems to be connected to back-end storage components e.g. S3).

[adwk67]

I see from the source code that the code path is determined by these variables: which values (or defaults) do you have set for them?

AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_PATH
AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD
AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_COMPRESSION

[maxgruber19]

Our xcom is fully stored in s3, no matter how big the xcom is. We do that to keep the footprint of the corresponding postgres as low as possible. Other reason is to avoid getting confused because some are stored in pg and some in s3.

We will try reproducing the error with your example dag, I'll come back here when we're done with that.

stackable@airflow-webserver-default-0 /stackable $ export | grep XCOM
declare -x AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_PATH="s3://hpo_default@int-da-airflow/xcom"
declare -x AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD="0"
declare -x AIRFLOW__CORE__XCOM_BACKEND="airflow.providers.common.io.xcom.backend.XComObjectStorageBackend"

[adwk67]

You could maybe also try setting AIRFLOW__CORE__EXECUTE_TASKS_NEW_PYTHON_INTERPRETER to True, which will introduce a performance hit, but should get around caching of S3 connection objects.