fix(core): recognize new duplicate-name error wording; add actionable train() errors

sagemaker-core/src/sagemaker/core/common_utils.py

@@ -2446,6 +2446,42 @@ def _check_job_status(job, desc, status_key_name):
         )
 
 
+# Error codes and message patterns the service returns when a create call collides
+# with an existing resource. The service message wording has changed over time
+# (e.g. CreatePipeline/CreateExperiment now return "... names must be unique within
+# an AWS account ..." instead of "... already exists"), so match every known variant.
+# The uniqueness pattern deliberately includes the "within an AWS account" scope so
+# that other uniqueness validation errors (e.g. duplicate step names WITHIN a
+# pipeline definition) are not mistaken for a resource-name collision.
+_ALREADY_EXISTS_ERROR_CODES = ("ValidationException", "ResourceInUse")
+_ALREADY_EXISTS_MSG_PATTERNS = (
+    "Cannot create already existing",
+    "already exists",
+    "must be unique within an AWS account",
+)
+
+
+def _is_resource_already_exists_error(error) -> bool:
+    """Check whether a botocore ClientError means "this resource already exists".
+
+    Use this predicate for every load-or-create / upsert flow instead of matching
+    a single hardcoded message substring, so that service message wording changes
+    do not silently break the already-exists branch.
+
+    Args:
+        error (botocore.exceptions.ClientError): The error raised by a create call.
+
+    Returns:
+        bool: True if the error indicates a name collision with an existing resource.
+    """
+    error_response = getattr(error, "response", None) or {}
+    error_code = error_response.get("Error", {}).get("Code", "")
+    error_message = error_response.get("Error", {}).get("Message", "")
+    return error_code in _ALREADY_EXISTS_ERROR_CODES and any(
+        pattern in error_message for pattern in _ALREADY_EXISTS_MSG_PATTERNS
+    )
+
+
 def _create_resource(create_fn):
     """Call create function and accepts/pass when resource already exists.
 
@@ -2462,14 +2498,7 @@ def _create_resource(create_fn):
         # create function succeeded, resource does not exist already
         return True
     except ClientError as ce:
-        error_code = ce.response["Error"]["Code"]
-        error_message = ce.response["Error"]["Message"]
-        already_exists_exceptions = ["ValidationException", "ResourceInUse"]
-        already_exists_msg_patterns = ["Cannot create already existing", "already exists"]
-        if not (
-            error_code in already_exists_exceptions
-            and any(p in error_message for p in already_exists_msg_patterns)
-        ):
+        if not _is_resource_already_exists_error(ce):
             raise ce
         # no new resource created as resource already exists
         return False

sagemaker-core/src/sagemaker/core/experiments/experiment.py

@@ -20,7 +20,7 @@
 from sagemaker.core.apiutils import _base_types
 from sagemaker.core.experiments.trial import _Trial
 from sagemaker.core.experiments.trial_component import _TrialComponent
-from sagemaker.core.common_utils import format_tags
+from sagemaker.core.common_utils import format_tags, _is_resource_already_exists_error
 from sagemaker.core.telemetry.telemetry_logging import _telemetry_emitter
 from sagemaker.core.telemetry.constants import Feature
 
@@ -169,9 +169,7 @@ def _load_or_create(
                 sagemaker_session=sagemaker_session,
             )
         except ClientError as ce:
-            error_code = ce.response["Error"]["Code"]
-            error_message = ce.response["Error"]["Message"]
-            if not (error_code == "ValidationException" and "already exists" in error_message):
+            if not _is_resource_already_exists_error(ce):
                 raise ce
             # already exists
             experiment = Experiment.load(experiment_name, sagemaker_session)

sagemaker-core/src/sagemaker/core/experiments/trial.py

@@ -18,7 +18,7 @@
 from sagemaker.core.apiutils import _base_types
 import sagemaker.core.experiments._api_types as _api_types
 from sagemaker.core.experiments.trial_component import _TrialComponent
-from sagemaker.core.common_utils import format_tags
+from sagemaker.core.common_utils import format_tags, _is_resource_already_exists_error
 
 
 class _Trial(_base_types.Record):
@@ -280,9 +280,7 @@ def _load_or_create(
                 sagemaker_session=sagemaker_session,
             )
         except ClientError as ce:
-            error_code = ce.response["Error"]["Code"]
-            error_message = ce.response["Error"]["Message"]
-            if not (error_code == "ValidationException" and "already exists" in error_message):
+            if not _is_resource_already_exists_error(ce):
                 raise ce
             # already exists
             trial = _Trial.load(trial_name, sagemaker_session)

sagemaker-core/src/sagemaker/core/experiments/trial_component.py

@@ -20,7 +20,7 @@
 from sagemaker.core.apiutils import _base_types
 import sagemaker.core.experiments._api_types as _api_types
 from sagemaker.core.experiments._api_types import TrialComponentSearchResult
-from sagemaker.core.common_utils import format_tags
+from sagemaker.core.common_utils import format_tags, _is_resource_already_exists_error
 
 
 class _TrialComponent(_base_types.Record):
@@ -338,9 +338,7 @@ def _load_or_create(
                 sagemaker_session=sagemaker_session,
             )
         except ClientError as ce:
-            error_code = ce.response["Error"]["Code"]
-            error_message = ce.response["Error"]["Message"]
-            if not (error_code == "ValidationException" and "already exists" in error_message):
+            if not _is_resource_already_exists_error(ce):
                 raise ce
             # already exists
             run_tc = _TrialComponent.load(trial_component_name, sagemaker_session)

sagemaker-core/tests/unit/experiments/test_load_or_create.py

@@ -0,0 +1,83 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+#     http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+"""Regression tests for the load-or-create flows in the experiments modules.
+
+The service changed its duplicate-name error wording from '... already exists'
+to '... names must be unique within an AWS account ...'. These tests pin the
+already-exists branch against BOTH wordings so a future wording change cannot
+silently turn load-or-create back into a hard failure.
+"""
+from __future__ import absolute_import
+
+import pytest
+from unittest.mock import patch
+from botocore.exceptions import ClientError
+
+from sagemaker.core.experiments.experiment import Experiment
+from sagemaker.core.experiments.trial import _Trial
+from sagemaker.core.experiments.trial_component import _TrialComponent
+
+
+LEGACY_MESSAGE = "Experiment exp-1 already exists"
+NEW_MESSAGE = "Experiment names must be unique within an AWS account and region"
+
+
+def _validation_error(message):
+    return ClientError(
+        {"Error": {"Code": "ValidationException", "Message": message}}, "create"
+    )
+
+
+@pytest.mark.parametrize("message", [LEGACY_MESSAGE, NEW_MESSAGE])
+def test_experiment_load_or_create_loads_on_duplicate_name(message):
+    with patch.object(Experiment, "create", side_effect=_validation_error(message)):
+        with patch.object(Experiment, "load") as mock_load:
+            result = Experiment._load_or_create(experiment_name="exp-1")
+
+    mock_load.assert_called_once_with("exp-1", None)
+    assert result is mock_load.return_value
+
+
+def test_experiment_load_or_create_reraises_unrelated_validation_error():
+    with patch.object(
+        Experiment, "create", side_effect=_validation_error("1 validation error detected")
+    ):
+        with patch.object(Experiment, "load") as mock_load:
+            with pytest.raises(ClientError):
+                Experiment._load_or_create(experiment_name="exp-1")
+
+    mock_load.assert_not_called()
+
+
+@pytest.mark.parametrize("message", [LEGACY_MESSAGE, NEW_MESSAGE])
+def test_trial_load_or_create_loads_on_duplicate_name(message):
+    with patch.object(_Trial, "create", side_effect=_validation_error(message)):
+        with patch.object(_Trial, "load") as mock_load:
+            mock_load.return_value.experiment_name = "exp-1"
+            result = _Trial._load_or_create(experiment_name="exp-1", trial_name="trial-1")
+
+    mock_load.assert_called_once_with("trial-1", None)
+    assert result is mock_load.return_value
+
+
+@pytest.mark.parametrize("message", [LEGACY_MESSAGE, NEW_MESSAGE])
+def test_trial_component_load_or_create_loads_on_duplicate_name(message):
+    with patch.object(_TrialComponent, "create", side_effect=_validation_error(message)):
+        with patch.object(_TrialComponent, "load") as mock_load:
+            result, is_existed = _TrialComponent._load_or_create(
+                trial_component_name="tc-1"
+            )
+
+    mock_load.assert_called_once_with("tc-1", None)
+    assert result is mock_load.return_value
+    assert is_existed is True

sagemaker-core/tests/unit/test_common_utils.py

@@ -2911,3 +2911,95 @@ def test_save_to_default_bucket_preserves_kms(self, tmp_path):
         assert merged["ServerSideEncryption"] == "aws:kms"
         assert merged["SSEKMSKeyId"] == "kms-key-id"
         assert merged["ExpectedBucketOwner"] == "111111111111"
+
+
+class TestIsResourceAlreadyExistsError:
+    """Test the shared already-exists predicate used by load-or-create/upsert flows."""
+
+    @staticmethod
+    def _client_error(code, message):
+        from botocore.exceptions import ClientError
+
+        return ClientError({"Error": {"Code": code, "Message": message}}, "create_pipeline")
+
+    def test_matches_legacy_already_exists_message(self):
+        from sagemaker.core.common_utils import _is_resource_already_exists_error
+
+        error = self._client_error("ValidationException", "Pipeline pipe-1 already exists")
+        assert _is_resource_already_exists_error(error) is True
+
+    def test_matches_new_names_must_be_unique_message(self):
+        """The service now returns 'names must be unique' instead of 'already exists'."""
+        from sagemaker.core.common_utils import _is_resource_already_exists_error
+
+        error = self._client_error(
+            "ValidationException",
+            "Pipeline names must be unique within an AWS account and region",
+        )
+        assert _is_resource_already_exists_error(error) is True
+
+    def test_matches_experiment_names_must_be_unique_message(self):
+        from sagemaker.core.common_utils import _is_resource_already_exists_error
+
+        error = self._client_error(
+            "ValidationException",
+            "Experiment names must be unique within an AWS account and region",
+        )
+        assert _is_resource_already_exists_error(error) is True
+
+    def test_matches_cannot_create_already_existing_message(self):
+        from sagemaker.core.common_utils import _is_resource_already_exists_error
+
+        error = self._client_error(
+            "ValidationException", "Cannot create already existing endpoint configuration"
+        )
+        assert _is_resource_already_exists_error(error) is True
+
+    def test_matches_resource_in_use_code(self):
+        from sagemaker.core.common_utils import _is_resource_already_exists_error
+
+        error = self._client_error(
+            "ResourceInUse", "Job name must be unique within an AWS account and region"
+        )
+        assert _is_resource_already_exists_error(error) is True
+
+    def test_rejects_other_validation_errors(self):
+        from sagemaker.core.common_utils import _is_resource_already_exists_error
+
+        error = self._client_error("ValidationException", "1 validation error detected")
+        assert _is_resource_already_exists_error(error) is False
+
+    def test_rejects_uniqueness_errors_scoped_within_a_resource(self):
+        """Uniqueness violations INSIDE a definition (e.g. duplicate step names) are
+        not name collisions -- treating them as already-exists would make upsert()
+        wrongly fall through to update() and mask the real validation error."""
+        from sagemaker.core.common_utils import _is_resource_already_exists_error
+
+        error = self._client_error(
+            "ValidationException", "Step names must be unique within a pipeline"
+        )
+        assert _is_resource_already_exists_error(error) is False
+
+    def test_rejects_other_error_codes(self):
+        from sagemaker.core.common_utils import _is_resource_already_exists_error
+
+        error = self._client_error("ResourceLimitExceeded", "names must be unique")
+        assert _is_resource_already_exists_error(error) is False
+
+    def test_create_resource_accepts_names_must_be_unique(self):
+        """_create_resource treats the new service wording as already-exists."""
+        from botocore.exceptions import ClientError
+        from sagemaker.core.common_utils import _create_resource
+
+        def _raise():
+            raise ClientError(
+                {
+                    "Error": {
+                        "Code": "ValidationException",
+                        "Message": "Pipeline names must be unique within an AWS account",
+                    }
+                },
+                "create_pipeline",
+            )
+
+        assert _create_resource(_raise) is False

sagemaker-mlops/src/sagemaker/mlops/workflow/pipeline.py

@@ -40,6 +40,7 @@
     retry_with_backoff,
     format_tags,
     Tags,
+    _is_resource_already_exists_error,
 )
 
 # Orchestration imports (now in mlops)
@@ -180,6 +181,8 @@ def create(
         description: str = None,
         tags: Optional[Tags] = None,
         parallelism_config: ParallelismConfiguration = None,
+        *,
+        _log_name_collision_hint: bool = True,
     ) -> Dict[str, Any]:
         """Creates a Pipeline in the Pipelines service.
 
@@ -219,7 +222,21 @@ def create(
             Tags=tags,
         )
         # TODO: replace with sagemaker-core methods
-        return self.sagemaker_session.sagemaker_client.create_pipeline(**kwargs)
+        try:
+            return self.sagemaker_session.sagemaker_client.create_pipeline(**kwargs)
+        except ClientError as ce:
+            # upsert() handles the name collision itself (create-or-update), so it
+            # suppresses this hint -- otherwise every successful upsert of an
+            # existing pipeline would log a misleading ERROR.
+            if _log_name_collision_hint and _is_resource_already_exists_error(ce):
+                logger.error(
+                    "A pipeline named '%s' already exists in this account and Region. "
+                    "To update the existing pipeline (or create it only if missing), call "
+                    "pipeline.upsert() instead of pipeline.create(). To keep both, choose a "
+                    "unique pipeline name.",
+                    self.name,
+                )
+            raise ce
 
     def _create_args(
         self, role_arn: str, description: str, parallelism_config: ParallelismConfiguration
@@ -353,11 +370,15 @@ def upsert(
             sagemaker_session=self.sagemaker_session,
         )
         try:
-            response = self.create(role_arn, description, tags, parallelism_config)
+            response = self.create(
+                role_arn,
+                description,
+                tags,
+                parallelism_config,
+                _log_name_collision_hint=False,
+            )
         except ClientError as ce:
-            error_code = ce.response["Error"]["Code"]
-            error_message = ce.response["Error"]["Message"]
-            if not (error_code == "ValidationException" and "already exists" in error_message):
+            if not _is_resource_already_exists_error(ce):
                 raise ce
             # already exists
             response = self.update(role_arn, description, parallelism_config=parallelism_config)