Fix: Log MWAA CLI stderr instead of raising an exception (#1527)

Author: izeigerman

sqlmesh/core/plan/evaluator.py

@@ -14,6 +14,7 @@
 Refer to `sqlmesh.core.plan`.
 """
 import abc
+import logging
 import typing as t
 
 from sqlmesh.core.console import Console, get_console
@@ -38,6 +39,8 @@
 from sqlmesh.utils.date import now
 from sqlmesh.utils.errors import SQLMeshError
 
+logger = logging.getLogger(__name__)
+
 
 class PlanEvaluator(abc.ABC):
     @abc.abstractmethod
@@ -370,7 +373,7 @@ def _apply_plan(self, plan: Plan, plan_request_id: str) -> None:
         )
 
         if stderr:
-            raise SQLMeshError(f"Failed to submit a plan application request:\n{stderr}")
+            logger.warning("MWAA CLI stderr:\n%s", stderr)
 
 
 def can_evaluate_before_promote(

tests/core/test_plan_evaluator.py

@@ -142,30 +142,6 @@ def test_mwaa_evaluator(sushi_plan: Plan, mocker: MockerFixture):
     mwaa_client_mock.wait_for_first_dag_run.assert_called_once()
 
 
-def test_mwaa_evaluator_error_from_cli(sushi_plan: Plan, mocker: MockerFixture):
-    mwaa_client_mock = mocker.Mock()
-    mwaa_client_mock.wait_for_dag_run_completion.return_value = True
-    mwaa_client_mock.wait_for_first_dag_run.return_value = "test_plan_application_dag_run_id"
-    mwaa_client_mock.set_variable.return_value = "", "Error"
-
-    state_sync_mock = mocker.Mock()
-
-    plan_dag_spec_json = """{"request_id": "test_request_id"}"""
-
-    plan_dag_spec_mock = mocker.Mock()
-    plan_dag_spec_mock.json.return_value = plan_dag_spec_json
-
-    create_plan_dag_spec_mock = mocker.patch("sqlmesh.schedulers.airflow.plan.create_plan_dag_spec")
-    create_plan_dag_spec_mock.return_value = plan_dag_spec_mock
-
-    evaluator = MWAAPlanEvaluator(mwaa_client_mock, state_sync_mock)
-
-    with pytest.raises(SQLMeshError, match=r"Failed to submit a plan application request:\nError"):
-        evaluator.evaluate(sushi_plan)
-
-    mwaa_client_mock.set_variable.assert_called_once_with(mocker.ANY, plan_dag_spec_json)
-
-
 def test_can_evaluate_before_promote(sushi_context: Context):
     parent_model_a = SqlModel(
         name="sushi.new_test_model_a",