test(feature-processor): Isolate pipeline names to fix flaky integ tests (#6095)

The feature processor to_pipeline integ tests hardcoded fixed pipeline
names (pipeline-name-01, pipeline-name-lf-01) and thus shared fixed S3
paths (s3://.../<pipeline_name>/function/payload.pkl). Combined with the
asymmetric-signing scheme introduced in PR #5816 (each to_pipeline call
generates a fresh ECDSA key pair, overwrites the signed payload, and
pins the matching public key into the pipeline env), concurrent CI
builds on the same account overwrite each other's payloads. A running
execution then verifies a payload signed by a different build's key,
producing DeserializationError: "Integrity check for the serialized
function or data failed" and pipeline execution status Failed.

Generate unique pipeline names via unique_name_from_base so each test
run and build uses an isolated S3 prefix, and restore cleanup_pipeline
in the finally blocks to avoid resource leakage.

Author: lucasjia-aws

sagemaker-mlops/tests/integ/feature_store/feature_processor/test_feature_processor_integ.py

@@ -33,6 +33,7 @@
 
 from tests.integ import DATA_DIR
 from sagemaker.core.helper.session_helper import Session, get_execution_role
+from sagemaker.core.common_utils import unique_name_from_base
 from sagemaker.core.s3 import S3Uploader
 from urllib.parse import urlparse
 from sagemaker.core.remote_function import remote
@@ -681,7 +682,7 @@ def test_to_pipeline_and_execute(
     pre_execution_commands,
     dependencies_path,
 ):
-    pipeline_name = "pipeline-name-01"
+    pipeline_name = unique_name_from_base("pipeline-name-01")
     car_data_feature_group_name = get_car_data_feature_group_name()
     car_data_aggregated_feature_group_name = get_car_data_aggregated_feature_group_name()
     try:
@@ -791,7 +792,7 @@ def transform(raw_s3_data_as_df):
         cleanup_feature_group(
             feature_groups["car_data_aggregated_feature_group"], sagemaker_session=sagemaker_session
         )
-        # cleanup_pipeline(pipeline_name="pipeline-name-01", sagemaker_session=sagemaker_session)
+        cleanup_pipeline(pipeline_name=pipeline_name, sagemaker_session=sagemaker_session)
 
 
 # @pytest.mark.skipif(
@@ -810,7 +811,7 @@ def test_to_pipeline_and_execute_with_lake_formation(
     pre_execution_commands,
     dependencies_path,
 ):
-    pipeline_name = "pipeline-name-lf-01"
+    pipeline_name = unique_name_from_base("pipeline-name-lf-01")
     car_data_feature_group_name = get_car_data_feature_group_name()
     car_data_fg = None
     try:
@@ -946,6 +947,7 @@ def transform(raw_s3_data_as_df):
             except Exception as e:
                 logging.warning(f"Failed to delete Glue table: {e}")
             cleanup_feature_group(car_data_fg, sagemaker_session=sagemaker_session)
+        cleanup_pipeline(pipeline_name=pipeline_name, sagemaker_session=sagemaker_session)
 
 
 # @pytest.mark.skipif(
@@ -959,7 +961,7 @@ def test_schedule_and_event_trigger(
     pre_execution_commands,
     dependencies_path,
 ):
-    pipeline_name = "pipeline-name-01"
+    pipeline_name = unique_name_from_base("pipeline-name-01")
     car_data_feature_group_name = get_car_data_feature_group_name()
     car_data_aggregated_feature_group_name = get_car_data_aggregated_feature_group_name()
     try:
@@ -1175,6 +1177,7 @@ def transform(raw_s3_data_as_df):
         cleanup_feature_group(
             feature_groups["car_data_aggregated_feature_group"], sagemaker_session=sagemaker_session
         )
+        cleanup_pipeline(pipeline_name=pipeline_name, sagemaker_session=sagemaker_session)
 
 
 def get_car_data_feature_group_name():