Prefect 1 0 gt 2 0 Migration Question Hi I have a flow that

Question

Prefect 1 0 gt 2 0 Migration Question Hi I have a flow that accepts lists of params as input The flow is composed of many tasks that occur sequentially and in parallel One of the initial tasks for a particular input param results in failure so the subsequent tasks are generally marked `Not Ready` with the `future result ` resolving to `None` The issue is that two of the final tasks of my flow accept futures of previous tasks as input and are listed in the `wait for` section as well These final two tasks instead are stuck in a `Pending` state and this causes the flow to hang instead of running to completion Any advice on how to properly implement the ideal behavior tasks that depend on futures from previous tasks that fail or didn t run should also fail not run and flow should run to completion would be greatly appreciated thank you In the logs I see the flow encountered an `UpstreamTaskError`

Mason Menges · Answer

Hey < Neil Natarajan> What version of prefect are you running for 2 0 specifically you should be able to implement a simple If check to ensure the data fits the format you re wanting to pass

Neil Natarajan · Answer

I am using 2 02 atm and this is a visual of what my flow looks like

Neil Natarajan · Answer

the last two tasks on the right are `Pending` and I see the following in my logs ```raise UpstreamTaskError prefect exceptions UpstreamTaskError Upstream task run 5ed91f00 90d8 4cd4 9e9c 2a6fcd3805ec did not reach a COMPLETED state ```

Neil Natarajan · Answer

The flow doesn t actually complete I have to terminate the process

Neil Natarajan · Answer

and then the flow is marked `Crashed`

Mason Menges · Answer

Well for starters you shouldn t normally need to call result on a task it should just return the output of the task do you have a minimum example of how you re flow tasks are setup to reproduce this with

Mason Menges · Answer

Do you see any of the logs for the flow run specifically around the failed tasks

Neil Natarajan · Answer

```2022 08 17 11 47 00 630 ERROR base events default exception handler 1738 Exception in callback Worker handle stimulus from task lt Task finishe ED state gt gt at Users neil natarajan PycharmProjects ssr manager venv lib python3 9 site packages distributed worker py 1844 handle lt Handle Worker handle stimulus from task lt Task finishe ED state gt gt at Users neil natarajan PycharmProjects ssr manager venv lib python3 9 site packages distributed worker py 1844 gt Traceback most recent call last File Users neil natarajan pyenv versions 3 9 10 lib python3 9 asyncio events py line 80 in run self context run self callback self args File Users neil natarajan PycharmProjects ssr manager venv lib python3 9 site packages distributed worker py line 193 in wrapper return method self args kwargs File Users neil natarajan PycharmProjects ssr manager venv lib python3 9 site packages distributed worker py line 1854 in handle stimulus from task super handle stimulus from task task File Users neil natarajan PycharmProjects ssr manager venv lib python3 9 site packages distributed worker state machine py line 3337 in handle stimulus from task stim = task result File Users neil natarajan PycharmProjects ssr manager venv lib python3 9 site packages distributed worker py line 180 in wrapper return await method self args kwargs type ignore File Users neil natarajan PycharmProjects ssr manager venv lib python3 9 site packages distributed worker py line 2186 in execute result = await apply function async File Users neil natarajan PycharmProjects ssr manager venv lib python3 9 site packages distributed worker py line 2938 in apply function async result = await function args kwargs File Users neil natarajan PycharmProjects ssr manager venv lib python3 9 site packages prefect engine py line 946 in begin task run return await orchestrate task run File Users neil natarajan PycharmProjects ssr manager venv lib python3 9 site packages prefect engine py line 1007 in orchestrate task run resolved parameters = await resolve inputs parameters File Users neil natarajan PycharmProjects ssr manager venv lib python3 9 site packages prefect engine py line 1230 in resolve inputs return await visit collection File Users neil natarajan PycharmProjects ssr manager venv lib python3 9 site packages prefect utilities collections py line 288 in visit collection values = await gather visit nested v for v in values File Users neil natarajan PycharmProjects ssr manager venv lib python3 9 site packages prefect utilities asyncutils py line 341 in gather keys append tg start soon call File Users neil natarajan PycharmProjects ssr manager venv lib python3 9 site packages prefect utilities asyncutils py line 317 in aexit retval = await self task group aexit tb File Users neil natarajan PycharmProjects ssr manager venv lib python3 9 site packages anyio backends asyncio py line 660 in aexit raise ExceptionGroup exceptions anyio backends asyncio ExceptionGroup 2 exceptions were raised in the task group ```

Mason Menges · Answer

Do you have an example of your flow code you d be able to share

Neil Natarajan · Answer

``` flow name= Create FD Instances Workflow task runner=DaskTaskRunner validate parameters=False def create fd instances workflow repo dict Dict str GitRepo create params list List CreateFDInstParams create fi name str create workflow uuids List str Beginning of Create Logic Preprocess Create Requests valid requests future = preprocess duplicate requests submit params list=create params list workflow uuids=create workflow uuids valid requests = valid requests future result raise on failure=False get fd type objects fd type objects futures = get fd type def submit git repo=repo dict ServiceRegistryRepo FUNCTIONAL DOMAIN DEFINITION value params=create params for create params in create params list validate that FI and FD Instance params are valid is fi fd valid futures = validate fi fd inputs submit wait for=fd type objects futures fi name=create fi name workflow uuid=create uuid valid request=valid req falcon inst repo=repo dict ServiceRegistryRepo FALCON INSTANCE DEFINITION value params=create params fd type=fd type obj for fd type obj create params create uuid valid req in zip fd type objects futures create params list create workflow uuids valid requests env name futures = get env name submit workflow uuid=c uuid fi name=create fi name falcon inst repo=repo dict ServiceRegistryRepo FALCON INSTANCE DEFINITION value params=c params for c params c uuid in zip create params list create workflow uuids determine if requested FI is stage prod req prod fi futures = requested prod fi submit wait for=env name futures workflow uuid=create workflow uuids env name=env name env git repo=repo dict ServiceRegistryRepo FALCON ENVIRONMENT DEFINITION value params=create params for env name create params in zip env name futures create params list create client to query gus simple sf client future = create simple sf client submit check for approval approvals futures = evaluate approval submit wait for= simple sf client future + req prod fi futures workflow uuid=c uuid fi name=create fi name params=c params client=simple sf client future req prod fi=req prod fi for c params c uuid req prod fi in zip create params list create workflow uuids req prod fi futures create base FD inst objects new created fd inst futures = create fd inst obj submit wait for=fd type objects futures params=c params obj uuid=c uuid fd type obj=fd obj for c params c uuid fd obj in zip create params list create workflow uuids fd type objects futures valid to write new fd futures = determine if new fd valid submit wait for= valid requests future + approvals futures + is fi fd valid futures + env name futures + fd type objects futures valid request=valid r approval=appr fi fd valid=v fi fd for appr v fi fd valid r in zip approvals futures is fi fd valid futures valid requests write objects locally for valid requests create success futures = write object submit wait for= valid requests future + new created fd inst futures + valid to write new fd futures + fd type objects futures + env name futures workflow uuid=c uuid valid req=valid to write new fd fi name=create fi name falcon inst repo=repo dict ServiceRegistryRepo FALCON INSTANCE DEFINITION value params=c params fd inst=new created fd inst workflow type=WorkflowTypesConfig CREATE FD NAME for c params c uuid valid to write new fd new created fd inst in zip create params list create workflow uuids valid to write new fd futures new created fd inst futures create results = future result raise on failure=False for future in create success futures ```

Mason Menges · Answer

At a glance It looks like you re calling submit on most of you re tasks you shouldn t need to do this anymore in recent versions of prefect this also applies to result have you tried running this without those set on your tasks or task results recent versions of prefect just return the resulting output of the task so you may need to adjust some of the checks you have in place to check for the output rather than the future

Neil Natarajan · Answer

Just to clarify what should I be calling instead of submit to execute the same task for many inputs in parallel

Neil Natarajan · Answer

< Mason Menges>

Neil Natarajan · Answer

also what version of Prefect were these changes made in I am following the prefect 2 0 docs

Mason Menges · Answer

Ah nope that s correct this was my bad I knew we had updated the tasks to return a python object instead of a future but I didn t know we had altered the task behavior to return a future when calling submit That said you should still be able to implement an if check on the result of the failing tasks to only run the downstream tasks if the result is not None

Mason Menges · Answer

this is a simple example that might be helpful ```from prefect import flow task task def taska num if num gt 5 return num + 1 else return None task def taskb some num return some num + 1 flow def flowa num inputs = 1 2 3 7 8 9 test futures = taska submit num=num val for num val in num inputs taskb test = taskb submit test future result for test future in test futures if test future result =None print taskb future result for taskb future in taskb test flowa ```

Neil Natarajan · Answer

Fair enough I ll try that

Neil Natarajan · Answer

thanks

Mason Menges · Answer

No problem smile

Neil Natarajan · Answer

So in my case I shouldn t pass the future and should pass the result directly

Mason Menges · Answer

I think that should work yes at least for validating the value itself otherwise passing the future is still fine I would think