diff --git a/airflow/providers/amazon/aws/utils/waiter_with_logging.py b/airflow/providers/amazon/aws/utils/waiter_with_logging.py
index 5f66b01cd2a78..1e506927a1f43 100644
--- a/airflow/providers/amazon/aws/utils/waiter_with_logging.py
+++ b/airflow/providers/amazon/aws/utils/waiter_with_logging.py
@@ -71,6 +71,7 @@ def wait(
             break
         except WaiterError as error:
             if "terminal failure" in str(error):
+                log.error("%s: %s", failure_message, _LazyStatusFormatter(status_args, error.last_response))
                 raise AirflowException(f"{failure_message}: {error}")
 
             log.info("%s: %s", status_message, _LazyStatusFormatter(status_args, error.last_response))
@@ -122,6 +123,7 @@ async def async_wait(
             break
         except WaiterError as error:
             if "terminal failure" in str(error):
+                log.error("%s: %s", failure_message, _LazyStatusFormatter(status_args, error.last_response))
                 raise AirflowException(f"{failure_message}: {error}")
 
             log.info("%s: %s", status_message, _LazyStatusFormatter(status_args, error.last_response))
diff --git a/tests/providers/amazon/aws/utils/test_waiter_with_logging.py b/tests/providers/amazon/aws/utils/test_waiter_with_logging.py
index 03427c2ac529a..2f753b84f3922 100644
--- a/tests/providers/amazon/aws/utils/test_waiter_with_logging.py
+++ b/tests/providers/amazon/aws/utils/test_waiter_with_logging.py
@@ -136,7 +136,7 @@ def test_wait_max_attempts_exceeded(self, mock_sleep, caplog):
             },
         )
 
-        mock_waiter.wait.call_count == 11
+        assert mock_waiter.wait.call_count == 2
         mock_sleep.assert_called_with(123)
         assert (
             caplog.record_tuples
@@ -165,6 +165,7 @@ def test_wait_with_failure(self, mock_sleep, caplog):
             last_response=generate_response("Failure"),
         )
         mock_waiter.wait.side_effect = [error, error, error, failure_error]
+
         with pytest.raises(AirflowException) as exc:
             wait(
                 waiter=mock_waiter,
@@ -175,6 +176,7 @@ def test_wait_with_failure(self, mock_sleep, caplog):
                 status_message="test status message",
                 status_args=["Status.State"],
             )
+
         assert "test failure message" in str(exc)
         mock_waiter.wait.assert_called_with(
             **{"test_arg": "test_value"},
@@ -183,17 +185,7 @@ def test_wait_with_failure(self, mock_sleep, caplog):
             },
         )
         assert mock_waiter.wait.call_count == 4
-        assert (
-            caplog.record_tuples
-            == [
-                (
-                    "airflow.providers.amazon.aws.utils.waiter_with_logging",
-                    logging.INFO,
-                    "test status message: Pending",
-                )
-            ]
-            * 3
-        )
+        assert caplog.messages == ["test status message: Pending"] * 3 + ["test failure message: Failure"]
 
     @mock.patch("time.sleep")
     def test_wait_with_list_response(self, mock_sleep, caplog):
@@ -279,7 +271,7 @@ def test_wait_with_incorrect_args(self, mock_sleep, caplog):
                 "MaxAttempts": 1,
             },
         )
-        mock_waiter.wait.call_count == 3
+        assert mock_waiter.wait.call_count == 3
         mock_sleep.assert_called_with(123)
         assert (
             caplog.record_tuples
@@ -320,7 +312,7 @@ def test_wait_with_multiple_args(self, mock_sleep, caplog):
             status_message="test status message",
             status_args=["Clusters[0].Status", "Clusters[0].StatusDetails", "Clusters[0].ClusterName"],
         )
-        mock_waiter.wait.call_count == 3
+        assert mock_waiter.wait.call_count == 3
         mock_sleep.assert_called_with(123)
         assert (
             caplog.record_tuples