fix(sdk): permit empty local execution outputs #localexecution (#10338)

sdk/python/kfp/local/executor_output_utils.py

@@ -49,15 +49,17 @@ def cast_protobuf_numbers(
     struct_pb2.Value to a dict/json, int will be upcast to float, even
     if the component output specifies int.
     """
-    output_parameter_types = [
+    int_output_keys = [
         output_param_name
         for output_param_name, parameter_spec in output_parameter_types.items()
         if parameter_spec.parameter_type ==
         pipeline_spec_pb2.ParameterType.ParameterTypeEnum.NUMBER_INTEGER
     ]
-    for float_output_key in output_parameter_types:
-        output_parameters[float_output_key] = int(
-            output_parameters[float_output_key])
+    for int_output_key in int_output_keys:
+        # avoid KeyError when the user never writes to the dsl.OutputPath
+        if int_output_key in output_parameters:
+            output_parameters[int_output_key] = int(
+                output_parameters[int_output_key])
     return output_parameters
 
 

sdk/python/kfp/local/subprocess_task_handler_test.py

@@ -404,6 +404,19 @@ def my_comp(out_param: dsl.OutputPath(str),) -> int:
         self.assertEqual(task.outputs['out_param'], 'HelloHello')
         self.assertEqual(task.outputs['Output'], 1)
 
+    def test_outputpath_result_not_written(self):
+        local.init(runner=local.SubprocessRunner(use_venv=True))
+
+        # use dsl.OutputPath(int) for more thorough testing
+        # want to ensure that the code that converts protobuf number to
+        # Python int permits unwritten outputs
+        @dsl.component
+        def my_comp(out_param: dsl.OutputPath(int)):
+            pass
+
+        task = my_comp()
+        self.assertEmpty(task.outputs)
+
 
 if __name__ == '__main__':
     unittest.main()