Warn for bad combination of parameters for RedshiftDataOperator (#69524)

Author: amoghrajesh

providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_data.py

@@ -125,6 +125,11 @@ def __init__(
         self.deferrable = deferrable
         self.session_id = session_id
         self.session_keep_alive_seconds = session_keep_alive_seconds
+        if self.deferrable and not self.wait_for_completion:
+            self.log.warning(
+                "deferrable=True and wait_for_completion=False are set; deferrable will be "
+                "ignored and this task will run non-deferrable."
+            )
 
     def execute(self, context: Context) -> list[GetStatementResultResponseTypeDef] | list[str]:
         """Execute a statement against Amazon Redshift."""

providers/amazon/tests/unit/amazon/aws/operators/test_redshift_data.py

@@ -87,6 +87,21 @@ def test_init(self):
         assert op.hook._verify is None
         assert op.hook._config is None
 
+    @mock.patch.object(RedshiftDataOperator, "log", new_callable=mock.MagicMock)
+    def test_init_warns_when_deferrable_has_no_effect(self, mock_log):
+        """deferrable=True is a no-op when wait_for_completion=False; the user should be told."""
+        RedshiftDataOperator(
+            task_id=TASK_ID,
+            database=DATABASE,
+            sql=SQL,
+            deferrable=True,
+            wait_for_completion=False,
+        )
+        mock_log.warning.assert_called_once_with(
+            "deferrable=True and wait_for_completion=False are set; deferrable will be "
+            "ignored and this task will run non-deferrable."
+        )
+
     @mock.patch("airflow.providers.amazon.aws.hooks.redshift_data.RedshiftDataHook.execute_query")
     @mock.patch("airflow.providers.amazon.aws.hooks.redshift_data.RedshiftDataHook.conn")
     def test_execute(self, mock_conn, mock_exec_query):