Remove unused semaphore/mutex pipeline config

Signed-off-by: sduvvuri1603 <[email protected]>

Author: sduvvuri1603

api/v2alpha1/go/pipelinespec/pipeline_spec.pb.go

@@ -1166,11 +1166,8 @@ message KubernetesWorkspaceConfig {
 
 // Spec for pipeline-level config options. See PipelineConfig DSL class.
 message PipelineConfig {
-  // Name of the semaphore key to control pipeline concurrency
-  string semaphore_key = 1;
-
-  // Name of the mutex to ensure mutual exclusion
-  string mutex_name = 2;
+  reserved 1, 2;
+  reserved "semaphore_key", "mutex_name";
 
   // Time to live configuration after the pipeline run is completed for
   // ephemeral resources created by the pipeline run.

api/v2alpha1/pipeline_spec.proto

@@ -306,9 +306,7 @@ var platformSpec = &specPB.PlatformSpec{
 				},
 			},
 			PipelineConfig: &specPB.PipelineConfig{
-				SemaphoreKey: "test-key",
-				MutexName:    "test-mutex",
-				ResourceTtl:  24,
+				ResourceTtl: 24,
 			},
 		},
 	},

backend/test/proto_tests/objects.go

@@ -15,8 +15,6 @@
         "project":  "test-project"
       },
       "pipelineConfig":  {
-        "semaphore_key":  "test-key",
-        "mutex_name":  "test-mutex",
         "resource_ttl":  24
       }
     }

backend/test/proto_tests/testdata/generated-1791485/platform_spec.json

@@ -1,7 +1,7 @@
 
-�

+x
 
-kubernetes�

+kubernetesj
 ;
 9
 root-executor(
@@ -12,6 +12,4 @@ kubernetes
 test-image
 kubernetes
 
-projecttest-project"
-test-key
-test-mutex
+projecttest-project"

backend/test/proto_tests/testdata/generated-1791485/platform_spec.pb

@@ -388,16 +388,17 @@ index d986a048b..866696419 100644
 +  google.protobuf.Struct pvc_spec_patch = 1;
 +}
 +
- // Spec for pipeline-level config options. See PipelineConfig DSL class.
+// Spec for pipeline-level config options. See PipelineConfig DSL class.
  message PipelineConfig {
-   // Name of the semaphore key to control pipeline concurrency
-@@ -1115,4 +1128,7 @@ message PipelineConfig {
+   reserved 1, 2;
+   reserved "semaphore_key", "mutex_name";
+
    // Time to live configuration after the pipeline run is completed for
    // ephemeral resources created by the pipeline run.
    int32 resource_ttl = 3;
-+
-+  // Configuration for the workspace
-+  optional WorkspaceConfig workspace = 4;
+
+   // Configuration for the workspace
+   optional WorkspaceConfig workspace = 4;
  }
 ```
 

proposals/11875-pipeline-workspace/README.md

@@ -4334,68 +4334,6 @@ def my_pipeline():
                 compiler.Compiler().compile(
                     pipeline_func=my_pipeline, package_path=output_yaml)
 
-
-class TestPipelineSemaphoreMutex(unittest.TestCase):
-
-    def test_pipeline_with_semaphore(self):
-        """Test that pipeline config correctly sets the semaphore key."""
-        config = PipelineConfig()
-        config.semaphore_key = 'semaphore'
-
-        @dsl.pipeline(pipeline_config=config)
-        def my_pipeline():
-            task = comp()
-
-        with tempfile.TemporaryDirectory() as tempdir:
-            output_yaml = os.path.join(tempdir, 'pipeline.yaml')
-            compiler.Compiler().compile(
-                pipeline_func=my_pipeline, package_path=output_yaml)
-
-            with open(output_yaml, 'r') as f:
-                pipeline_docs = list(yaml.safe_load_all(f))
-
-        platform_spec = None
-        for doc in pipeline_docs:
-            if 'platforms' in doc:
-                platform_spec = doc
-                break
-
-        self.assertIsNotNone(platform_spec,
-                             'No platforms section found in compiled output')
-        kubernetes_spec = platform_spec['platforms']['kubernetes'][
-            'pipelineConfig']
-        self.assertEqual(kubernetes_spec['semaphoreKey'], 'semaphore')
-
-    def test_pipeline_with_mutex(self):
-        """Test that pipeline config correctly sets the mutex name."""
-        config = PipelineConfig()
-        config.mutex_name = 'mutex'
-
-        @dsl.pipeline(pipeline_config=config)
-        def my_pipeline():
-            task = comp()
-
-        with tempfile.TemporaryDirectory() as tempdir:
-            output_yaml = os.path.join(tempdir, 'pipeline.yaml')
-            compiler.Compiler().compile(
-                pipeline_func=my_pipeline, package_path=output_yaml)
-
-            with open(output_yaml, 'r') as f:
-                pipeline_docs = list(yaml.safe_load_all(f))
-
-        platform_spec = None
-        for doc in pipeline_docs:
-            if 'platforms' in doc:
-                platform_spec = doc
-                break
-
-        self.assertIsNotNone(platform_spec,
-                             'No platforms section found in compiled output')
-        kubernetes_spec = platform_spec['platforms']['kubernetes'][
-            'pipelineConfig']
-        self.assertEqual(kubernetes_spec['mutexName'], 'mutex')
-
-
 class ExtractInputOutputDescription(unittest.TestCase):
 
     def test_no_descriptions(self):

sdk/python/kfp/compiler/compiler_test.py

@@ -2253,11 +2253,6 @@ def _merge_pipeline_config(pipelineConfig: pipeline_config.PipelineConfig,
     if workspace is not None:
         config_dict['workspace'] = workspace.get_workspace()
 
-    if pipelineConfig.semaphore_key is not None:
-        config_dict['semaphoreKey'] = pipelineConfig.semaphore_key
-    if pipelineConfig.mutex_name is not None:
-        config_dict['mutexName'] = pipelineConfig.mutex_name
-
     if config_dict:
         json_format.ParseDict({'pipelineConfig': config_dict},
                               platformSpec.platforms['kubernetes'])

sdk/python/kfp/compiler/pipeline_spec_builder.py

@@ -96,57 +96,5 @@ def set_kubernetes_config(self,
 class PipelineConfig:
     """PipelineConfig contains pipeline-level config options."""
 
-    def __init__(self,
-                 workspace: Optional[WorkspaceConfig] = None,
-                 semaphore_key: Optional[str] = None,
-                 mutex_name: Optional[str] = None):
+    def __init__(self, workspace: Optional[WorkspaceConfig] = None):
         self.workspace = workspace
-        self._semaphore_key = semaphore_key
-        self._mutex_name = mutex_name
-
-    @property
-    def semaphore_key(self) -> Optional[str]:
-        """Get the semaphore key for controlling pipeline concurrency.
-
-        Returns:
-            Optional[str]: The semaphore key, or None if not set.
-        """
-        return self._semaphore_key
-
-    @semaphore_key.setter
-    def semaphore_key(self, value: str):
-        """Set the semaphore key to control pipeline concurrency.
-
-        Pipelines with the same semaphore key will be limited to a configured maximum
-        number of concurrent executions. This allows you to control resource usage by
-        ensuring that only a specific number of pipelines can run simultaneously.
-
-        Note: A pipeline can use both semaphores and mutexes together. The pipeline
-        will wait until all required locks are available before starting.
-
-        Args:
-            value (str): The semaphore key name for controlling concurrent executions.
-        """
-        self._semaphore_key = (value and value.strip()) or None
-
-    @property
-    def mutex_name(self) -> Optional[str]:
-        """Get the mutex name for exclusive pipeline execution.
-
-        Returns:
-            Optional[str]: The mutex name, or None if not set.
-        """
-        return self._mutex_name
-
-    @mutex_name.setter
-    def mutex_name(self, value: str):
-        """Set the name of the mutex to ensure mutual exclusion.
-
-        Pipelines with the same mutex name will only run one at a time. This ensures
-        exclusive access to shared resources and prevents conflicts when multiple
-        pipelines would otherwise compete for the same resources.
-
-        Args:
-            value (str): Name of the mutex for exclusive pipeline execution.
-        """
-        self._mutex_name = (value and value.strip()) or None

sdk/python/kfp/dsl/pipeline_config.py

@@ -211,8 +211,6 @@
     my_pipeline as retry_pipeline
 from test_data.sdk_compiled_pipelines.valid.pipeline_with_secret_as_volume import \
     pipeline_secret_volume
-from test_data.sdk_compiled_pipelines.valid.pipeline_with_semphore import \
-    pipeline_with_semaphore
 from test_data.sdk_compiled_pipelines.valid.pipeline_with_string_machine_fields_pipeline_input import \
     pipeline as pipeline_with_string_machine_fields_pipeline_input
 from test_data.sdk_compiled_pipelines.valid.pipeline_with_string_machine_fields_task_output import \
@@ -1002,13 +1000,6 @@ def __repr__(self) -> str:
                 compiled_file_name='pythonic_artifact_with_single_return.yaml',
                 expected_compiled_file_path=f'{_VALID_PIPELINE_FILES}/pythonic_artifact_with_single_return.yaml'
             ),
-            TestData(
-                pipeline_name='pipeline-with-semaphore',
-                pipeline_func=pipeline_with_semaphore,
-                pipline_func_args=None,
-                compiled_file_name='pipeline_with_semaphore.yaml',
-                expected_compiled_file_path=f'{_VALID_PIPELINE_FILES}/pipeline_with_semaphore.yaml'
-            ),
             TestData(
                 pipeline_name='pipeline-with-condition-dynamic-task-output',
                 pipeline_func=pipeline_with_dynamic_condition_output,