[Bug Fix] Delete ResourceOp should not have output parameters (#1822)

* Fix bug where delete resource op should not have success_condition, failure_condition, and output parameters

* remove unnecessary whitespace

* compiler test for delete resource ops should retrieve templates from spec instead of root

sdk/python/kfp/dsl/_resource_op.py

@@ -105,6 +105,10 @@ def __init__(self,
 
         if merge_strategy and action != "apply":
             ValueError("You can't set merge_strategy when action != 'apply'")
+
+        # if action is delete, there should not be any outputs, success_condition, and failure_condition
+        if action == "delete" and (success_condition or failure_condition or attribute_outputs):
+            ValueError("You can't set success_condition, failure_condition, or attribute_outputs when action == 'delete'")
 
         init_resource = {
             "action": action,
@@ -117,6 +121,13 @@ def __init__(self,
 
         self.k8s_resource = k8s_resource
 
+        # if action is delete, there should not be any outputs, success_condition, and failure_condition
+        if action == "delete":
+            self.attribute_outputs = {}
+            self.outputs = {}
+            self.output = None
+            return
+
         # Set attribute_outputs
         extra_attribute_outputs = \
             attribute_outputs if attribute_outputs else {}

sdk/python/tests/compiler/compiler_tests.py

@@ -633,3 +633,35 @@ def init_container_pipeline():
         init_container = init_containers[0]
         self.assertEqual(init_container, {'image':'alpine:latest', 'command': ['echo', 'bye'], 'name': 'echo'})
 
+
+  def test_delete_resource_op(self):
+      """Test a pipeline with a delete resource operation."""
+      from kubernetes import client as k8s
+
+      @dsl.pipeline()
+      def some_pipeline():
+        # create config map object with k6 load test script
+        config_map = k8s.V1ConfigMap(
+          api_version="v1",
+          data={"foo": "bar"},
+          kind="ConfigMap",
+          metadata=k8s.V1ObjectMeta(
+              name="foo-bar-cm",
+              namespace="default"
+          )
+        )        
+        # delete the config map in k8s
+        dsl.ResourceOp(
+          name="delete-config-map", 
+          action="delete", 
+          k8s_resource=config_map
+        )
+
+      workflow_dict = kfp.compiler.Compiler()._compile(some_pipeline)
+      delete_op_template = [template for template in workflow_dict['spec']['templates'] if template['name'] == 'delete-config-map'][0]
+
+      # delete resource operation should not have success condition, failure condition or output parameters.
+      # See https://github.com/argoproj/argo/blob/5331fc02e257266a4a5887dfe6277e5a0b42e7fc/cmd/argoexec/commands/resource.go#L30
+      self.assertIsNone(delete_op_template.get("successCondition"))
+      self.assertIsNone(delete_op_template.get("failureCondition"))
+      self.assertDictEqual(delete_op_template.get("outputs"), {})