[mohamedzeidan2021]

Issue

Fixes #6495

QualityCheckStep.arguments and ClarifyCheckStep.arguments build their CreateProcessingJob request by hand, and the hand-rolled shape is wrong in two places:

1. KmsKeyId is placed inside ProcessingS3Output. The API does not accept it there — it belongs on ProcessingOutputConfig, where it applies to every output.
2. check_job_config.tags is never read, so tags configured on CheckJobConfig are silently dropped.

Validating the pre-fix request against the CreateProcessingJob service model (botocore.validate.ParamValidator) reports:

Unknown parameter in ProcessingOutputConfig.Outputs[0].S3Output: "KmsKeyId",
must be one of: S3Uri, LocalPath, S3UploadMode

Both are v2 → v3 regressions. In v2 the request came from Session._get_process_request via Processor._get_process_args, which set output_config["KmsKeyId"] and request["Tags"] = processor.tags. The v3 hand-rolled builders lost both.

ClarifyCheckStep is affected identically — the issue reports QualityCheckStep, but the two arguments properties are the same code, so the same request is malformed for every check step.

Fix

Move KmsKeyId onto ProcessingOutputConfig and add Tags from check_job_config.tags (omitted when unset or empty, rather than sent empty).

The two builders were identical apart from the inputs, the output and the default job name, so the ~70 duplicated lines move into a shared _build_check_job_request helper in a new private sagemaker/mlops/workflow/_check_job_request.py. The fix then lands once and the two steps can no longer drift. Each step keeps its own trim_request_dict call, since that is about the pipeline definition rather than the processing request.

Before:

{
  "ProcessingOutputConfig": {
    "Outputs": [{"S3Output": {"KmsKeyId": "arn:aws:kms:..."}}]
  }
}

After:

{
  "ProcessingOutputConfig": {
    "Outputs": [{"S3Output": {"S3Uri": "...", "LocalPath": "...", "S3UploadMode": "EndOfJob"}}],
    "KmsKeyId": "arn:aws:kms:..."
  },
  "Tags": [{"Key": "test", "Value": "quality-check"}]
}

No behavior other than the two fields changes: every other field name, default (VolumeSizeInGB 30, MaxRuntimeInSeconds 86400) and optional-field condition is carried over unchanged, including the getattr fallbacks.

Testing

Reproduced the issue first with the script from #6495 (offline, mocked boto session): pre-fix ProcessingOutputConfig.KmsKeyId is None, Outputs[0].S3Output.KmsKeyId is set and Tags is absent. Post-fix that inverts.

• New sagemaker-mlops/tests/unit/workflow/test_check_job_request.py (11 tests) covers the shared builder, including a guard that validates the produced request against the real CreateProcessingJob input shape via botocore.validate.ParamValidator — the pre-fix shape fails this check, so the regression cannot come back silently. Also covers tag omission when unset/empty, inputs with no s3_input, empty inputs, the default job name fallback, the optional processor fields and the VolumeSizeInGB/MaxRuntimeInSeconds defaults.
• Updated test_quality_check_step_kms.py and test_clarify_check_step_kms.py: these pinned the incorrect S3Output.KmsKeyId placement, so they now assert ProcessingOutputConfig.KmsKeyId and that S3Output has no KmsKeyId, plus new Tags propagation classes.
• Updated tests/integ/test_check_step_kms_propagation.py for the corrected placement and added tag coverage.

Against pre-fix source the updated unit tests fail 12 / pass 12; with the fix all pass. Full sagemaker-mlops/tests/unit/workflow suite: 412 passed. black (100) clean, flake8 (120) clean, pylint 10.00/10 on the new module, pydocstyle clean.

Backwards compatibility

The emitted request becomes valid where it was previously rejected by the API, and tags configured on CheckJobConfig now reach the job as documented. Pipeline definitions that previously carried S3Output.KmsKeyId will serialize it under ProcessingOutputConfig on their next upsert; no public API, signature or default changes.

[github-actions]

🤖 AI code review

This PR fixes two v2→v3 regressions in the check steps' hand-built CreateProcessingJob request: it moves KmsKeyId from ProcessingS3Output (where the API rejects it) to ProcessingOutputConfig, and propagates CheckJobConfig.tags as Tags. The ~70 duplicated lines in QualityCheckStep.arguments and ClarifyCheckStep.arguments are extracted into a shared _build_check_job_request helper. I verified CheckJobConfig.tags exists (normalized via format_tags into the correct [{"Key","Value"}] shape), that all imports (RequestType, ProcessingInput, ProcessingOutput) resolve at the base commit, and that the refactor preserves the original None/empty-input handling and getattr defaults. The change matches the guidance in issue #6495 and is well covered by new/updated unit tests (including a ParamValidator guard against the service model) and integ tests. No issues found.

_{Reviewed commit 3924572. Automated review; verify before acting.}