### Apache Airflow version

2.7.3

### What happened

when the trigger is running the execute_complete and the status is not success getting error

```
    def execute_complete(self, context, event=None):
        if event["status"] != "success":
            raise AirflowException(f"Error while running job: {event}")
        else:
            self.log.info(event["message"])

ip-10-5-55-118.eu-central-1.compute.internal
*** Reading remote log from Cloudwatch log_group: airflow-tf-stg-euc1-airflow-analytics-engine-Task log_stream: dag_id=execute_model/run_id=manual__2023-11-26T12_53_43.787114+00_00/task_id=submit_emr_task/attempt=1.log.
[2023-11-26, 12:58:32 UTC] {{taskinstance.py:1937}} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/models/baseoperator.py", line 1606, in resume_execution
    return execute_callable(context)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/providers/amazon/aws/operators/emr.py", line 618, in execute_complete
    self.log.info("%s", event["message"])
                        ~~~~~^^^^^^^^^^^
KeyError: 'message'
[2023-11-26, 12:58:32 UTC] {{standard_task_runner.py:104}} ERROR - Failed to execute job 2858894 for task submit_emr_task ('message'; 2194)
```

### What you think should happen instead

not getting the rror

### How to reproduce

run EmrContainerSensor with deferrable true on aws

### Operating System

aws

### Versions of Apache Airflow Providers

apache-airflow-providers-amazon==8.4.0
apache-airflow-providers-apache-kafka==1.2.0
apache-airflow-providers-common-sql==1.6.1
apache-airflow-providers-ftp==3.4.2
apache-airflow-providers-http==4.5.0
apache-airflow-providers-imap==3.2.2
apache-airflow-providers-postgres==5.6.0
apache-airflow-providers-sqlite==3.4.3


### Deployment

Amazon (AWS) MWAA

### Deployment details

_No response_

### Anything else

_No response_

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

"""
        Make an API call with boto3 and get cluster-level details.
        .. seealso::
            https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/emr.html#EMR.Client.describe_cluster
        :return: response
        """
        emr_client = self.hook.conn
        self.log.info("Poking cluster %s", self.job_flow_id)
        response = emr_client.describe_cluster(ClusterId=self.job_flow_id)
        EmrClusterLink.persist(
            context=context,
            operator=self,
            region_name=self.hook.conn_region_name,
            aws_partition=self.hook.conn_partition,
            job_flow_id=self.job_flow_id,
        )
        EmrLogsLink.persist(
            context=context,
            operator=self,
            region_name=self.hook.conn_region_name,
            aws_partition=self.hook.conn_partition,
            job_flow_id=self.job_flow_id,
            log_uri=get_log_uri(cluster=response),
        )
        return response
    @staticmethod
    def state_from_response(response: dict[str, Any]) -> str: