Scheduler not updating package files

Question

I'm developing a DAG on Cloud Composer; my code is separated into a main python file and one package with subfolders, it looks like this:

my_dag1.py
package1/__init__.py
package1/functions.py
package1/package2/__init__.py
package1/package2/more_functions.py

I updated one of the functions on package1/functions.py to take an additional argument (and update the reference in my_dag1.py). The code would run correctly on my local environment and I was not getting any errors when running

gcloud beta composer environments run my-airflow-environment list_dags --location europe-west1

But the Web UI raised a python error

TypeError: my_function() got an unexpected keyword argument 'new_argument'

I have tried to rename the function and the error changed to NameError: name 'my_function' is not defined
I tried changing the name of the DAG and to upload the files to the dag folder zipped and unzipped, but nothing worked.

The error disappeared only after I renamed the package folder.

I suspect the issue is related to the scheduler picking up my_dag1.py but not package1/functions.py. The error appeared out of nowhere as I have made similar updates on the previous weeks.

Any idea on how to fix this issue without refactoring the whole code structure?

EDIT-1

Here's the link to related discussion on Google Groups

The imports are using this syntax: from package1.functions import my_function and they work without issues in my local Airflow environment (and in Cloud Composer before I added the attribute to the function) — TAPeri
I experienced some issues like this on Cloud Composer (but not on local environment), and this is really a pain to debug. Did you try to delete your files (gcloud beta composer environments storage dags delete --environment my-airflow-environment --location europe-west1 <your directory>) and reupload them? It can also be an issue on the Web UI, but there is no way to restart the webserver on Cloud Composer. As a last resort, can you try to upload your files on a freshly created Cloud Composer environment and see if that works? That could help to isolate your problem. — norbjd
I deleted all the files using the Cloud Storage web interface and that didn't help. I'll try the composer console command next time I come across the issue, and see if that makes any difference. Thanks for the suggestion! — TAPeri

Kun Wang Kun Wang · Accepted Answer · 2019-01-14T23:57:11

I've run into a similar issue. the "Broken DAG" error won't dismiss in Web UI. I guess this is a cache bug in Web server of AirFlow.

Background.
1. I created a customized operator with Airflow Plugin features.
2. After I import the customized operator, the airflow Web UI keep shows the Broken DAG error says that it can't find the customized operator.
Why I think it's a bug in Web server Airflow?
1. I can manually run the DAG with the command airflow test, so the import should be correct.
2. Even if I remove the related DAG file from the /dags/ folder of airflow, the error still there.
Here are What I did to resolve this issue.
1. restart airflow web service. (sometimes you can resolve the issue only by this).
2. make sure no DAG is running, restart airflow scheduler service.
3. make sure no DAG is running, restart airflow worker

Hopefully, it can help someone has the same issue.

Scheduler not updating package files

2 Answers