Fix issue where tasks preceeding parallel for loops that recieve pipe…

…line parameters, are wrongly expected to have task attributes

sdk/python/kfp/compiler/compiler_utils.py

@@ -762,7 +762,7 @@ def get_dependencies(
                 # then make this validation dsl.Collected-aware
                 elif isinstance(upstream_parent_group, tasks_group.ParallelFor):
                     upstream_tasks_that_downstream_consumers_from = [
-                        channel.task.name for channel in task._channel_inputs
+                        channel.task.name for channel in task._channel_inputs if channel.task
                     ]
                     has_data_exchange = upstream_task.name in upstream_tasks_that_downstream_consumers_from
                     # don't raise for .after