chore(sdk): v2 compiler to throw no task defined error (#6545)

* chore(sdk): v2 compiler to throw  no task defined error

* update and clean up release notes

* Update RELEASE.md

* Update RELEASE.md

sdk/RELEASE.md

@@ -12,6 +12,8 @@
 
 ## Bug Fixes and Other Changes
 
+* v2 compiler to throw no task defined error. [\#6545](https://github.com/kubeflow/pipelines/pull/6545)
+
 ## Documentation Updates
 
 
@@ -23,12 +25,20 @@
 
 ## Breaking Changes
 
+* N/A
+
 ### For Pipeline Authors
 
+* N/A
+
 ### For Component Authors
 
+* N/A
+
 ## Deprecations
 
+* N/A
+
 ## Bug Fixes and Other Changes
 
 * Define PipelineParameterChannel and PipelineArtifactChannel in v2. [\#6470](https://github.com/kubeflow/pipelines/pull/6470)
@@ -42,6 +52,9 @@
 
 ## Documentation Updates
 
+* N/A
+
+
 # 1.8.0
 
 ## Major Features and Improvements
@@ -55,10 +68,16 @@
 
 ### For Pipeline Authors
 
+* N/A
+
 ### For Component Authors
 
+* N/A
+
 ## Deprecations
 
+* N/A
+
 ## Bug Fixes and Other Changes
 
 * Fix bug in PodSpec that overwrites nodeSelector [\#6512](https://github.com/kubeflow/pipelines/issues/6512)
@@ -72,6 +91,8 @@
 
 ## Documentation Updates
 
+* N/A
+
 # 1.7.2
 
 ## Major Features and Improvements

sdk/python/kfp/v2/compiler/compiler.py

@@ -1107,6 +1107,9 @@ class Compiler(object):
         with dsl.Pipeline(pipeline_name) as dsl_pipeline:
             pipeline_func(*args_list)
 
+        if not dsl_pipeline.ops:
+            raise ValueError('Task is missing from pipeline.')
+
         self._validate_exit_handler(dsl_pipeline)
         self._sanitize_and_inject_artifact(dsl_pipeline)
 

sdk/python/kfp/v2/compiler/compiler_test.py

@@ -24,6 +24,22 @@ from kfp.v2 import dsl
 from kfp.dsl import types
 
 
+VALID_PRODUCER_COMPONENT_SAMPLE = components.load_component_from_text("""
+    name: producer
+    inputs:
+    - {name: input_param, type: String}
+    outputs:
+    - {name: output_model, type: Model}
+    - {name: output_value, type: Integer}
+    implementation:
+      container:
+        image: gcr.io/my-project/my-image:tag
+        args:
+        - {inputValue: input_param}
+        - {outputPath: output_model}
+        - {outputPath: output_value}
+    """)
+
 class CompilerTest(unittest.TestCase):
 
     def test_compile_simple_pipeline(self):
@@ -155,6 +171,17 @@ class CompilerTest(unittest.TestCase):
             compiler.Compiler().compile(
                 pipeline_func=my_pipeline, package_path='output.json')
 
+    def test_compile_pipeline_with_missing_task_should_raise_error(self):
+
+        @dsl.pipeline(name='test-pipeline', pipeline_root='dummy_root')
+        def my_pipeline(text: str):
+            pass
+
+        with self.assertRaisesRegex(
+                ValueError,'Task is missing from pipeline.'):
+            compiler.Compiler().compile(
+                  pipeline_func=my_pipeline, package_path='output.json')
+
     def test_compile_pipeline_with_misused_inputuri_should_raise_error(self):
 
         component_op = components.load_component_from_text("""
@@ -204,7 +231,7 @@ class CompilerTest(unittest.TestCase):
     def test_compile_pipeline_with_invalid_name_should_raise_error(self):
 
         def my_pipeline():
-            pass
+            VALID_PRODUCER_COMPONENT_SAMPLE(input_param='input')
 
         with self.assertRaisesRegex(
                 ValueError,
@@ -220,7 +247,7 @@ class CompilerTest(unittest.TestCase):
 
             @dsl.pipeline(name='test-pipeline', pipeline_root='gs://path')
             def my_pipeline():
-                pass
+                VALID_PRODUCER_COMPONENT_SAMPLE(input_param='input')
 
             target_json_file = os.path.join(tmpdir, 'result.json')
             compiler.Compiler().compile(