[rsareddy0329]

Issue #, if available: Relates to #5796

Description of changes:

Processor.run / FrameworkProcessor.run waited via self.latest_job.wait(). The
generated ProcessingJob.wait() → refresh() resolves a default, global SageMaker
client (Base.get_sagemaker_client()), not the region of the sagemaker_session that
created the job. With a non-default-region session, the wait/describe and log tailing hit
the wrong region and fail — the reported bug (#5796). (V2 is unaffected: it waits through
self.sagemaker_session.logs_for_processing_job / wait_for_processing_job.)

This routes the wait path through the session:

• logs=True → logs_for_processing_job(sagemaker_session, ...), which already resolves
  the logs client from the session. It's also fixed to describe the job via
  sagemaker_session.sagemaker_client (region-aware) instead of
  ProcessingJob.get(...).refresh() — the chained .refresh() re-described through the
  global client (wrong region) and read PascalCase keys (ProcessingJobStatus) off a
  snake_case resource __dict__.
• logs=False → new _wait_for_processing_job(sagemaker_session, job_name), which polls
  describe_processing_job through the session's region-aware client and raises on failure.

Testing:

• test_run_with_wait (updated): wait=True, logs=False goes through _wait_for_processing_job
  with the processor's session (no longer latest_job.wait()).
• test_run_with_wait_and_logs_uses_session: logs=True routes through logs_for_processing_job
  with the session.
• test_wait_for_processing_job_uses_session_region / _raises_on_failure: the helper describes
  via sagemaker_session.sagemaker_client and raises on a Failed status.
• Full test_processing.py (125 passed) and model_monitor suites (82 passed, uses
  logs_for_processing_job) green. flake8 clean on changed files.

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[github-actions]

Claude finished @rsareddy0329's task in 1m 39s —— View job

---

Code Review — fix(processing): respect session region when waiting for processing jobs

• Read PR diff and changed files
• Trace helpers (_wait_until, _check_job_status, logs_for_processing_job) and callers
• Assess correctness, tests, and conventions
• Post review

---

Summary

This is a solid, well-targeted fix. The root cause is correctly diagnosed: latest_job.wait() → refresh() resolves the global Base.get_sagemaker_client() rather than the region of the sagemaker_session that created the job, so a non-default-region session hits the wrong region (#5796). Routing the wait path through sagemaker_session.sagemaker_client is the right fix and matches how v2 behaved.

I also want to call out a second, real bug this incidentally fixes: the old logs_for_processing_job read description["ProcessingJobStatus"] off ProcessingJob.get(...).refresh().__dict__, which is a snake_case resource dict — that key would never exist, so the logs path would have raised KeyError. Switching to describe_processing_job (PascalCase service response) makes _get_initial_job_state/_check_job_status work correctly. Good catch.

The split of latest_job.wait(logs=logs) into logs=True → logs_for_processing_job(..., wait=True) and logs=False → _wait_for_processing_job(...) preserves the prior semantics (log tailing + terminal-status check vs. plain poll + status check), and _describe_if_terminal returning None until a terminal state correctly drives _wait_until.

No correctness or security concerns found. The changes are backward compatible (logs_for_processing_job signature unchanged; _wait_for_processing_job is new) and well covered by the four added/updated tests.

Minor suggestions (non-blocking nits)

1. Docstring raise type is inaccurate. _wait_for_processing_job (and the pre-existing logs_for_processing_job) document Raises: ValueError, but _check_job_status actually raises exceptions.UnexpectedStatusException / exceptions.CapacityError (common_utils.py:2445). Worth correcting the new docstring to reflect the real exception type.

2. Test assertion is broad. test_wait_for_processing_job_raises_on_failure uses pytest.raises(Exception), which would pass on any exception (including an unrelated KeyError). Asserting the specific type (UnexpectedStatusException) would make the test more meaningful.

3. Duplicated block. The identical 11-line if logs: ... else: ... wait block now appears in both run methods (processing.py:363 and :941). This mirrors the pre-existing duplication, so it's fine to leave, but extracting a small self._wait_for_latest_job(wait, logs) helper would reduce drift risk if this logic changes again.

4. Uncovered terminal state. "Stopped" is handled by _check_job_status (warns, doesn't raise) but isn't exercised by a test; optional to add.

Nice work — the fix is correct and the tests target the regressed behavior precisely.
· fix/processing-wait-region