Keep poke_interval/max_attempts as deprecated SageMakerTrigger aliases

Preserve backward compatibility for SageMakerTrigger after switching to the
AwsBaseWaiterTrigger naming: poke_interval and max_attempts are still accepted
as deprecated aliases for waiter_delay and waiter_max_attempts, emitting
AirflowProviderDeprecationWarning. This keeps existing keyword callers working
and lets deferred-task triggers serialized by older versions deserialize after
upgrade.

Author: o-nikolas

providers/amazon/src/airflow/providers/amazon/aws/triggers/sagemaker.py

@@ -18,13 +18,15 @@
 from __future__ import annotations
 
 import asyncio
+import warnings
 from collections import Counter
 from collections.abc import AsyncIterator
 from enum import IntEnum
 from typing import TYPE_CHECKING
 
 from botocore.exceptions import WaiterError
 
+from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.providers.amazon.aws.hooks.sagemaker import SageMakerHook
 from airflow.providers.amazon.aws.triggers.base import AwsBaseWaiterTrigger
 from airflow.providers.common.compat.sdk import AirflowException
@@ -46,6 +48,8 @@ class SageMakerTrigger(AwsBaseWaiterTrigger):
     :param region_name: The AWS region where the job is running. Used to build the hook.
     :param verify: Whether or not to verify SSL certificates. Used to build the hook.
     :param botocore_config: Configuration dictionary for the botocore client. Used to build the hook.
+    :param poke_interval: (deprecated) use ``waiter_delay`` instead.
+    :param max_attempts: (deprecated) use ``waiter_max_attempts`` instead.
     """
 
     def __init__(
@@ -58,7 +62,25 @@ def __init__(
         region_name: str | None = None,
         verify: bool | str | None = None,
         botocore_config: dict | None = None,
+        poke_interval: int | None = None,
+        max_attempts: int | None = None,
     ):
+        if poke_interval is not None:
+            warnings.warn(
+                "`poke_interval` is deprecated and will be removed in a future release. "
+                "Please use `waiter_delay` instead.",
+                AirflowProviderDeprecationWarning,
+                stacklevel=2,
+            )
+            waiter_delay = poke_interval
+        if max_attempts is not None:
+            warnings.warn(
+                "`max_attempts` is deprecated and will be removed in a future release. "
+                "Please use `waiter_max_attempts` instead.",
+                AirflowProviderDeprecationWarning,
+                stacklevel=2,
+            )
+            waiter_max_attempts = max_attempts
         self.job_name = job_name
         self.job_type = job_type
         super().__init__(

providers/amazon/tests/unit/amazon/aws/triggers/test_sagemaker.py

@@ -22,6 +22,7 @@
 import pytest
 from botocore.exceptions import WaiterError
 
+from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.providers.amazon.aws.hooks.sagemaker import SageMakerHook
 from airflow.providers.amazon.aws.triggers.sagemaker import SageMakerPipelineTrigger, SageMakerTrigger
 from airflow.triggers.base import TriggerEvent
@@ -54,6 +55,23 @@ def test_sagemaker_trigger_serialize(self):
         assert args["aws_conn_id"] == AWS_CONN_ID
         assert args["region_name"] == REGION_NAME
 
+    @pytest.mark.parametrize(
+        ("deprecated_kwarg", "canonical_attr", "value"),
+        [
+            ("poke_interval", "waiter_delay", 17),
+            ("max_attempts", "attempts", 21),
+        ],
+    )
+    def test_sagemaker_trigger_deprecated_params(self, deprecated_kwarg, canonical_attr, value):
+        with pytest.warns(AirflowProviderDeprecationWarning, match=deprecated_kwarg):
+            trigger = SageMakerTrigger(
+                job_name=JOB_NAME,
+                job_type=JOB_TYPE,
+                aws_conn_id=AWS_CONN_ID,
+                **{deprecated_kwarg: value},
+            )
+        assert getattr(trigger, canonical_attr) == value
+
     def test_sagemaker_trigger_hook_uses_generic_params(self):
         sagemaker_trigger = SageMakerTrigger(
             job_name=JOB_NAME,