Why is my Airflow hanging up if I send a http request inside a task?

System: MacOS Apple M1 (local machine)

Airflow: 2.5.3

Executor: Local with Postgres Database

I am trying to implement some externally triggered workflows which load data from our REST APIs. I am using the Python Operator to run the code and trigger the process manually using the Airflow UI. However, when the execution gets to the task which has code sending a http request, it hangs forever and the laptop starts running really hot.

The task is defined in another file and I am importing it as a module. Here are the contents of the task file (tasks/import_logs.py)

import requests

def import_logs(**context):
    print("[Sasha] Running log importer")
    context["ti"].xcom_push(
        key="logs", value=["log/location/1", "log/location/2"])
    print("Log locations pushed to xcom")
    # Define the URL for the dummy endpoint
    url = 'https://jsonplaceholder.typicode.com/posts'

    # Define the payload for the JSON request
    payload = {
        "title": "foo",
        "body": "bar",
        "userId": 1
    }

    # Define the headers for the request
    headers = {'Content-Type': 'application/json'}

    # Send the POST request to the dummy endpoint
    response = requests.post(url, json=payload, headers=headers)

    # Print the response status code and content
    print(f'Response status code: {response.status_code}')
    print(f'Response content: {response.content}')

Here is the Dag definition:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
from tasks.import_logs import import_logs
from tasks.import_tops import import_tops
from tasks.process_input import process_input
from tasks.process_log_data import process_logs
from tasks.output_logs import output_logs
from tasks.cleanup import cleanup
from tasks.trigger_data_update import trigger_data_update


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 3, 31),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG('process_log', default_args=default_args, schedule_interval=None)

validate_input = PythonOperator(
    task_id='validate_input',
    python_callable=process_input,
    provide_context=True,
    dag=dag
)

import_log = PythonOperator(
    task_id='import_logs',
    python_callable=import_logs,
    provide_context=True,
    dag=dag
)

import_top = PythonOperator(
    task_id='import_tops',
    python_callable=import_tops,
    provide_context=True,
    dag=dag
)

process_log = PythonOperator(
    task_id='process_logs',
    python_callable=process_logs,
    provide_context=True,
    dag=dag
)

output_log = PythonOperator(
    task_id='write_logs',
    python_callable=output_logs,
    provide_context=True,
    dag=dag
)

cleanup_task = PythonOperator(
    task_id='cleanup',
    python_callable=cleanup,
    provide_context=True,
    dag=dag
)

update_task = PythonOperator(
    task_id='trigger_data_update',
    python_callable=trigger_data_update,
    provide_context=True,
    dag=dag
)

validate_input >> [import_top, import_log] >> process_log >> output_log >> [cleanup_task, update_task]

if __name__ == "__main__":
    import json
    with open('test_conf/process_log.json', 'r') as f:
        conf = json.load(f)
    dag.test(
        run_conf=conf
    )

It started happening with the Sequential executor. Since then I have moved the instance to a Local executor, stripped all the extra code and just left a simple dummy request to try and narrow down the error, but it just keeps happening.

I also tried using the Airflow HTTP Hook with a Connection, which led to the same behavior. Reading about it it seems that sending a request from a task should not be an issue, so I abandoned that approach.

At this point all other tasks are dummy as well, just printing things. Running the test at the bottom of the DAG file goes through OK without any issues. I am thinking to run Airflow in debug mode and start debugging, but that will kill a lot of time.

I also encounter http request hanging out problem just like you.

My local environment is airflow2.5.1/python3.8.13/M1 macOS.

It seems like a macOS python package problem.

I fixed it by adding env like this below:

export NO_PROXY="*"

https://github.com/apache/airflow/discussions/24463

Hope this can help you.

Switching from PythonOperator to PythonVirtualenvOperator solved this for me. However, it added a lot of new issues related communication between tasks due to some objects not being serializable.