be explicit about which job run logs are being retrieved

anyscale_provider/hooks/anyscale.py

@@ -137,11 +137,11 @@ def terminate_service(self, service_name: str, time_delay: int) -> bool:
             raise AirflowException(f"Service termination failed with error: {e}")
         return True
 
-    def get_job_logs(self, job_id: str) -> str:
+    def get_job_logs(self, job_id: str, run: str | None) -> str:
         """
          Fetch the logs for a job.
 
         :param job_id: Required. The ID of the job.
         """
-        logs: str = self.client.job.get_logs(id=job_id)
+        logs: str = self.client.job.get_logs(id=job_id, run=run)
         return logs

anyscale_provider/triggers/anyscale.py

@@ -59,8 +59,11 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
                 await asyncio.sleep(self.poll_interval)
 
             # Fetch and print logs
+            job_status = self.hook.get_job_status(self.job_id)
             loop = asyncio.get_running_loop()
-            logs = await loop.run_in_executor(None, partial(self.hook.get_job_logs, job_id=self.job_id))
+            logs = await loop.run_in_executor(
+                None, partial(self.hook.get_job_logs, job_id=self.job_id, run=job_status.runs[-1].name)
+            )
             for log in logs.split("\n"):
                 self.log.info(log)