Long running lambda wait for response times out in Airflow

We invoke certain lambdas via an Airflow operator. In the operator, a call to lambda.invoke in the boto3 library is made (via the AwsLambdaHook).

However, one of the lambdas has a long running time (10-15 mins). Initially the operator received a boto3 timeout exception due to the fact that the default timeout on invoke is 60 seconds.

So we extended the AwsLambdaHook to allow for configuration to be passed to the boto client (code for reference):

from airflow.contrib.hooks.aws_hook import AwsHook


class AwsLambdaHook(AwsHook):
    """
    Interact with AWS Lambda

    :param function_name: AWS Lambda Function Name
    :type function_name: str
    :param region_name: AWS Region Name (example: us-west-2)
    :type region_name: str
    :param log_type: Tail Invocation Request
    :type log_type: str
    :param qualifier: AWS Lambda Function Version or Alias Name
    :type qualifier: str
    :param invocation_type: AWS Lambda Invocation Type (RequestResponse, Event etc)
    :type invocation_type: str
    :param config: Config to pass to the lambda connection
    :type config: boto3.Config
    """

    def __init__(
        self,
        function_name,
        region_name=None,
        log_type="None",
        qualifier="$LATEST",
        invocation_type="RequestResponse",
        config=None,
        *args,
        **kwargs
    ):
        self.function_name = function_name
        self.region_name = region_name
        self.log_type = log_type
        self.invocation_type = invocation_type
        self.qualifier = qualifier
        self.config = config
        super(AwsLambdaHook, self).__init__(*args, **kwargs)

    def get_conn(self):
        self.conn = self.get_client_type("lambda", self.region_name, self.config)
        return self.conn

    def invoke_lambda(self, payload):
        """
        Invoke Lambda Function
        """

        awslambda_conn = self.get_conn()

        response = awslambda_conn.invoke(
            FunctionName=self.function_name,
            InvocationType=self.invocation_type,
            LogType=self.log_type,
            Payload=payload,
            Qualifier=self.qualifier,
        )

        return response

And we pass the following config:

from botocore.client import Config

config_dict = {"connect_timeout": 5, "read_timeout": 900}
config = Config(**config_dict)

Now, the operator no longer recieves a boto3 timeout exception. However, it ended up hitting our Airflow task execution timeout (configured to 1 hour). In the logs I saw the following:

[2019-10-25 07:52:05,172] {aws_lambda_operator.py:114} INFO - AWS Lambda: invoking Lambdas20
[2019-10-25 07:52:05,316] {logging_mixin.py:95} INFO - [[34m2019-10-25 07:52:05,316[0m] {[34mconnectionpool.py:[0m735} INFO[0m - Starting new HTTPS connection (1): lambda.ap-southeast-1.amazonaws.com[0m
[2019-10-25 08:07:07,097] {logging_mixin.py:95} INFO - [[34m2019-10-25 08:07:07,097[0m] {[34mconnectionpool.py:[0m735} INFO[0m - Starting new HTTPS connection (2): lambda.ap-southeast-1.amazonaws.com[0m
[2019-10-25 08:22:08,489] {logging_mixin.py:95} INFO - [[34m2019-10-25 08:22:08,489[0m] {[34mconnectionpool.py:[0m735} INFO[0m - Starting new HTTPS connection (3): lambda.ap-southeast-1.amazonaws.com[0m
[2019-10-25 08:37:11,402] {logging_mixin.py:95} INFO - [[34m2019-10-25 08:37:11,402[0m] {[34mconnectionpool.py:[0m735} INFO[0m - Starting new HTTPS connection (4): lambda.ap-southeast-1.amazonaws.com[0m
[2019-10-25 08:52:05,172] {logging_mixin.py:95} INFO - [[34m2019-10-25 08:52:05,172[0m] {[34mtime

It seems as if there is never a response from AWS. The lambda has already completed though. The same lambda code (just working on less data and running in a different region) does work fine with the Lambda operator, so I am pretty sure the problem is not with the lambda.

Now I have 2 questions:

1. Has anyone experienced similar behaviour before of invoking a lambda before? What was the resolution?

2. Is there an alternative approach where we can asynchronously poll the result? I looked in the boto3 documentation, but it looked like invoke_async was deprecated.