HttpHook: Unusable response object for slower REST API call

[Avphy]

Apache Airflow version:
AIRFLOW_VERSION=1.10.12

Kubernetes version (if you are using kubernetes) (use kubectl version):
NA

Environment:

• OS (e.g. from /etc/os-release): Oracle Linux
• Airflow Setup: Single node, CeleryExecutor setup (Airflow, DB, Celery worker running in same docker, Only Redis in saperate docker)

What happened:
When REST API call takes a longer time (~2 mins in my example) to return response. Response validation in HttpHook class doesn't work (literally no trace in logs, after response.raise_for_status() call in check_response() method)

Same api call works fine when called with below python script

import requests
import logging as log
from airflow import AirflowException
url = 'http://localhost:8085/example/taskrun'
response = requests.put(url, headers={"Content-Type": "application/json", "di-internal-request": "true"})
try:
    log.info("Response validation")
    response.raise_for_status()
    # print(response.text)
except requests.exceptions.Timeout as exc:
    log.error("HTTP Timeout error: %s", exc.response.text)
    log.error(exc.response.reason)
    raise AirflowException(str(exc.response.status_code) + ":" + exc.response.text)
except requests.exceptions.HTTPError as exc:
    log.error("HTTP error: %s", exc.response.text)
    log.error(exc.response.reason)
    raise AirflowException(str(exc.response.status_code) + ":" + exc.response.text)
except requests.exceptions.RequestException as exc:
    log.error("HTTP request error: %s", exc.response.text)
    log.error(exc.response.reason)
    raise AirflowException(str(exc.response.status_code) + ":" + exc.response.text)
print("API Call completed")   

Response
[2020-11-19 17:50:25,393] {api_test.py:10} INFO - Response validation
[2020-11-19 17:50:25,393] {api_test.py:18} ERROR - HTTP error: {"code" : "InternalServerError", "message" : there may be retries, if the retry attempt didn't exceed."}
[2020-11-19 17:50:25,395] {api_test.py:19} ERROR - Internal Server Error
Traceback (most recent call last):
File "/test/api_test.py", line 11, in <module>
 response.raise_for_status()
File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/requests/models.py", line 941, in raise_for_status
 raise HTTPError(http_error_msg, response=self)
requests.exceptions.HTTPError: 500 Server Error: Internal Server Error for url: http://localhost:8085/example/taskrun
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/test/api_test.py", line 20, in <module>
 raise AirflowException(str(exc.response.status_code) + ":" + exc.response.text)
airflow.exceptions.AirflowException: 500:{"code" : "InternalServerError", "message" : " there may be retries, if the retry attempt didn't exceed."}

What you expected to happen:
It should fall in below except block in HttpHook, and Response message/text/code should be printed

    def check_response(self, response):
        """
        Checks the status code and raise an AirflowException exception on non 2XX or 3XX
        status codes

        :param response: A requests response object
        :type response: requests.response
        """
        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError:
            self.log.error("HTTP error: %s", response.text)
            self.log.error(response.reason)
            raise AirflowException(str(response.status_code) + ":" + response.text)

How to reproduce it:
Create a DAG with HttpOperator and HttpSensor tasks. Call an API which returns response after delay (~2min). And Try to read the response object.

In my case I call api from on_success_callback/on_failure_callback method

**Note: We have created plugin to invoke Rest api call from within the callback methods.** 

Anything else we need to know:

In all our environments this behaviour is consistently reproducible.

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template!