LakeFS operators not working in task group

Issue

It seems that LakeFS operators are not working in Airflow TaskGroups

When using a LakeFS operator in a task group, the DAG disappear (not in error, but not listed by the scheduler). But if we remove the lakefs operator from a task group and use it directly in a DAG, the DAG magically appears.

It is a very strange behaviour (I've never seen that with Airflow), can be easily see locally with docker (see below)

Versions

Airflow 2.5.0
Lakefs Provider 0.44.0
Lakefs client 0.88.0
Python 3.7.15 (from public 2.5.0 Airflow image)

How to reproduce

Download official docker-compose.yaml from Airflow documentation

Create a folder dags with a file lakefs_issue_dag.py:


from airflow.decorators import dag, task_group
from airflow.operators.empty import EmptyOperator

import pendulum from lakefs_provider.operators.merge_operator import LakeFSMergeOperator

@task_group def my_group(): begin = EmptyOperator(task_id="begin")

# Comment below lines and the DAG will be scheduled
# Uncomment and it will disappear
merge = LakeFSMergeOperator(
    task_id="not_working_merge",
    lakefs_conn_id="conn",
    repo="repo",
    source_ref="source",
    destination_branch="dest",
    msg="msg",
)

begin >> merge

@dag( dag_id="lakefs_task_group_issue", schedule=None, start_date=pendulum.today('UTC').add(days=-1), ) def create_dag(arg1: str = "input"):

start = EmptyOperator(task_id="start")
group = my_group()

start >> group

working_merge = LakeFSMergeOperator(
    task_id="working_merge",
    lakefs_conn_id="conn",
    repo="repo",
    source_ref="source",
    destination_branch="dest",
    msg="msg",
)

start >> working_merge

dag = create_dag()


* Run

docker compose up airflow-init docker compose up

* Find _scheduler_ container id by running `docker ps` and then go on container by running

docker exec -it {container_id} bash

* Install lakefs provider

pip install airflow-provider-lakefs


* Now go on `localhost:8080`, connect with username/password `airflow` and see that DAG does not appear
* Run `airflow dags list` and `airflow dags list-import-errors`, we see nothing about dag `lakefs_task_group_issue`
* Comment DAG lines 13 to 22 and now see that the DAG appears (and can be found with `airflow dags list`
* Uncomment and the dags disappear again 
* Run `python dags/lakefs_issue_dag.py` and there are no errors (just a deprecation warning for Airflow 3)

treeverse / airflow-provider-lakeFS