chore(sdk): Warn on container component without command. Fixes #4800 (#6335)

* Warn on container component without command

* address cr comments

sdk/python/kfp/components/_components.py

@@ -23,6 +23,8 @@ import copy
 from collections import OrderedDict
 import pathlib
 from typing import Any, Callable, List, Mapping, NamedTuple, Sequence, Union
+import warnings
+
 from ._naming import _sanitize_file_name, _sanitize_python_function_name, generate_unique_name_conversion_table
 from ._yaml_utils import load_yaml
 from .structures import *
@@ -163,6 +165,17 @@ def _load_component_spec_from_component_text(text) -> ComponentSpec:
     component_dict = load_yaml(text)
     component_spec = ComponentSpec.from_dict(component_dict)
 
+    if isinstance(component_spec.implementation, ContainerImplementation) and (
+        component_spec.implementation.container.command is None):
+        warnings.warn(
+            'Container component must specify command to be compatible with KFP '
+            'v2 compatible mode and emissary executor, which will be the default'
+            ' executor for KFP v2.'
+            'https://www.kubeflow.org/docs/components/pipelines/installation/choose-executor/',
+            category=FutureWarning,
+        )
+
+
     # Calculating hash digest for the component
     import hashlib
     data = text if isinstance(text, bytes) else text.encode('utf-8')

sdk/python/kfp/components_tests/test_components.py

@@ -1043,6 +1043,21 @@ implementation:
         with self.assertRaises(TypeError):
             b_task = task_factory_b(in1=a_task.outputs['out1'])
 
+    def test_container_component_without_command_should_warn(self):
+        component_a = '''\
+name: component without command
+inputs:
+  - {name: in1, type: String}
+implementation:
+  container:
+    image: busybox
+'''
+
+        with self.assertWarnsRegex(
+            FutureWarning,
+            'Container component must specify command to be compatible with '
+            'KFP v2 compatible mode and emissary executor'):
+            task_factory_a = comp.load_component_from_text(component_a)
 
 if __name__ == '__main__':
     unittest.main()