diff --git a/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_data.py b/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_data.py index 3168b9f69c425..04dc8c3216297 100644 --- a/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_data.py +++ b/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.""" diff --git a/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_data.py b/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_data.py index 19182eb4c86b2..cb2f08db7e74c 100644 --- a/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_data.py +++ b/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):