Pass verify and botocore_config to BatchJobTrigger in BatchSensor (#72278)

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

@@ -102,6 +102,8 @@ def execute(self, context: Context) -> None:
                     job_id=self.job_id,
                     aws_conn_id=self.aws_conn_id,
                     region_name=self.region_name,
+		    verify=self.verify,
+		    botocore_config=self.botocore_config,
                     waiter_delay=int(self.poke_interval),
                     waiter_max_attempts=self.max_retries,
                 ),

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

@@ -92,6 +92,28 @@ def test_execute_in_deferrable_mode(self, deferrable_batch_sensor: BatchSensor):
             deferrable_batch_sensor.execute({})
         assert isinstance(exc.value.trigger, BatchJobTrigger), "Trigger is not a BatchJobTrigger"
 
+    def test_execute_in_deferrable_mode_passes_aws_configs(self):
+        """Asserts that verify and botocore_config survive into the serialized trigger payload."""
+        sensor = BatchSensor(
+            task_id="task",
+            job_id=JOB_ID,
+            region_name=AWS_REGION,
+            verify="/custom/ca_bundle.pem",
+            botocore_config={"read_timeout": 45},
+            deferrable=True,
+        )
+
+        with pytest.raises(TaskDeferred) as exc:
+            sensor.execute({})
+
+        trigger = exc.value.trigger
+        assert isinstance(trigger, BatchJobTrigger)
+
+        _, kwargs = trigger.serialize()
+        assert kwargs.get("region_name") == AWS_REGION
+        assert kwargs.get("verify") == "/custom/ca_bundle.pem"
+        assert kwargs.get("botocore_config") == {"read_timeout": 45}
+
     def test_execute_failure_in_deferrable_mode(self, deferrable_batch_sensor: BatchSensor):
         """Tests that an AirflowException is raised in case of error event"""