feat(sdk): SDK - Deprecation warning when using ContainerOp (#4166)

* SDK - Added warning when not using components

We have long advised our users to create reusable components.
Creating reusable components is as easy as creating ContainerOp instances, but the components are shareable, portable and are easier to support going forward.

* Disable warning for TFX

* Fixed the warning disabling logic

* Added tests

sdk/python/kfp/dsl/_component_bridge.py

@@ -42,6 +42,8 @@ def _create_container_op_from_component_and_arguments(
 
     container_spec = component_spec.implementation.container
 
+    old_warn_value = dsl.ContainerOp._DISABLE_REUSABLE_COMPONENT_WARNING
+    dsl.ContainerOp._DISABLE_REUSABLE_COMPONENT_WARNING = True
     task = dsl.ContainerOp(
         name=component_spec.name or _default_component_name,
         image=container_spec.image,
@@ -57,6 +59,7 @@ def _create_container_op_from_component_and_arguments(
             for input_name, path in resolved_cmd.input_paths.items()
         ],
     )
+    dsl.ContainerOp._DISABLE_REUSABLE_COMPONENT_WARNING = old_warn_value
 
     component_meta = copy.copy(component_spec)
     task._set_metadata(component_meta)

sdk/python/kfp/dsl/_container_op.py

@@ -960,6 +960,8 @@ def foo_pipeline(tag: str, pull_image_policy: str):
     # Excludes `file_outputs` and `outputs` as they are handled separately
     # in the compilation process to generate the DAGs and task io parameters.
 
+    _DISABLE_REUSABLE_COMPONENT_WARNING = False
+
     def __init__(
       self,
       name: str,
@@ -1010,6 +1012,19 @@ def __init__(
         """
 
         super().__init__(name=name, init_containers=init_containers, sidecars=sidecars, is_exit_handler=is_exit_handler)
+
+        if not ContainerOp._DISABLE_REUSABLE_COMPONENT_WARNING and '--component_launcher_class_path' not in (arguments or []):
+            # The warning is suppressed for pipelines created using the TFX SDK.
+            warnings.warn(
+                "Please create reusable components instead of constructing ContainerOp instances directly."
+                " Reusable components are shareable, portable and have compatibility and support guarantees."
+                " Please see the documentation: https://www.kubeflow.org/docs/pipelines/sdk/component-development/#writing-your-component-definition-file"
+                " The components can be created manually (or, in case of python, using kfp.components.create_component_from_func or func_to_container_op)"
+                " and then loaded using kfp.components.load_component_from_file, load_component_from_uri or load_component_from_text: "
+                "https://kubeflow-pipelines.readthedocs.io/en/latest/source/kfp.components.html#kfp.components.load_component_from_file",
+                category=DeprecationWarning,
+            )
+
         self.attrs_with_pipelineparams = BaseOp.attrs_with_pipelineparams + ['_container', 'artifact_arguments'] #Copying the BaseOp class variable!
 
         input_artifact_paths = {}

sdk/python/tests/dsl/component_bridge_tests.py

@@ -15,6 +15,7 @@
 import tempfile
 import textwrap
 import unittest
+import warnings
 import kfp
 from pathlib import Path
 from kfp.components import load_component_from_text, create_component_from_func
@@ -227,3 +228,18 @@ def test_converted_outputs(self):
 
         self.assertSetEqual(set(task1.outputs.keys()), {'Output 1', 'output_1'})
         self.assertIsNotNone(task1.output)
+
+    def test_reusable_component_warnings(self):
+        op1 = load_component_from_text('''\
+            implementation:
+                container:
+                    image: busybox
+            '''
+        )
+        with warnings.catch_warnings(record=True) as warning_messages:
+            op1()
+            deprecation_messages = list(str(message) for message in warning_messages if message.category == DeprecationWarning)
+            self.assertListEqual(deprecation_messages, [])
+
+        with self.assertWarnsRegex(DeprecationWarning, expected_regex='reusable'):
+            kfp.dsl.ContainerOp(name='name', image='image')