Pass the region_name from the GlueJobOperator / GlueJobSensor to the Trigger

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

@@ -249,6 +249,7 @@ def execute(self, context: Context):
                     aws_conn_id=self.aws_conn_id,
                     waiter_delay=self.waiter_delay,
                     waiter_max_attempts=self.waiter_max_attempts,
+                    region_name=self.region_name,
                 ),
                 method_name="execute_complete",
             )

providers/amazon/src/airflow/providers/amazon/aws/sensors/glue.py

@@ -105,6 +105,7 @@ def execute(self, context: Context) -> Any:
                     aws_conn_id=self.aws_conn_id,
                     waiter_delay=int(self.poke_interval),
                     waiter_max_attempts=self.max_retries,
+                    region_name=self.region_name,
                 ),
                 method_name="execute_complete",
             )

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

@@ -155,6 +155,11 @@ def test_execute_deferrable(self, _, mock_initialize_job):
 
         assert defer.value.trigger.job_name == JOB_NAME
         assert defer.value.trigger.run_id == JOB_RUN_ID
+        assert defer.value.trigger.region_name == "us-west-2"
+        assert not defer.value.trigger.verbose
+        assert defer.value.trigger.waiter_delay == 60
+        assert defer.value.trigger.attempts == 75
+        assert defer.value.trigger.aws_conn_id == "aws_default"
 
     @mock.patch.object(GlueJobHook, "print_job_logs")
     @mock.patch.object(GlueJobHook, "get_job_state")

providers/amazon/tests/unit/amazon/aws/sensors/test_glue.py

@@ -215,3 +215,28 @@ def test_custom_args(self):
         assert sensor.poke_interval == 10
         assert sensor.aws_conn_id == "custom_conn"
         assert sensor.max_retries == 20
+
+    def test_defferable_params_passed_to_trigger(self):
+        job_name = "job_name"
+        job_run_id = "job_run_id"
+        sensor = GlueJobSensor(
+            task_id="test_glue_job_sensor",
+            job_name=job_name,
+            run_id=job_run_id,
+            verbose=True,
+            deferrable=True,
+            poke_interval=10,
+            region_name="us-west-2",
+            aws_conn_id="custom_conn",
+            max_retries=20,
+        )
+        with pytest.raises(TaskDeferred) as defer:
+            sensor.execute({})
+
+        assert defer.value.trigger.job_name == job_name
+        assert defer.value.trigger.run_id == job_run_id
+        assert defer.value.trigger.region_name == "us-west-2"
+        assert defer.value.trigger.verbose
+        assert defer.value.trigger.waiter_delay == 10
+        assert defer.value.trigger.attempts == 20
+        assert defer.value.trigger.aws_conn_id == "custom_conn"