@Marvin In the following code snippet in Prefect 2, can I make sure flow does not return before the task completes?

@task(name='my_task', description='This will trigger a kubernetes spark job', retries=5, retry_delay_seconds=exponential_backoff(backoff_factor=600))
def my_task(processing_parameters):
    call_spark_end_point(processing_parameters)
    return

@flow(name='my_flow')
def my_flow(processing_parameters):
    my_task(processing_parameters)
    return

In Prefect, tasks are asynchronous by default. This means that when a task is called within a flow, the flow will not wait for the task to complete before moving on to the next task. However, in your case, it seems like you want the flow to wait for the

my_task

to complete before it returns. To achieve this, you can use Prefect's

wait_for_task

function. This function will block the flow until the specified task has completed. Here's how you can modify your code to use `wait_for_task`:

from prefect import task, flow, wait_for_task

@task(name='my_task', description='This will trigger a kubernetes spark job', retries=5, retry_delay_seconds=exponential_backoff(backoff_factor=600))
def my_task(processing_parameters):
    call_spark_end_point(processing_parameters)
    return

@flow(name='my_flow')
def my_flow(processing_parameters):
    task_result = my_task(processing_parameters)
    wait_for_task(task_result)
    return

In this modified code,

wait_for_task

is called with the result of

my_task

. This will cause the flow to block until

my_task

has completed. Please note that

wait_for_task

is a blocking operation, and it will prevent the flow from doing anything else until the specified task has completed. If you have other tasks that can be run in parallel, you might want to consider running them before you call

wait_for_task

.

wait_for_task is not a thing, marvin is wrong, task calls will block by default (unless you .submit them), the flow will not return until the tasks/subflows it calls are done (again, unless you submit them)

Thanks yeah was looking for

wait_for_task

🙃

however (not sure if relevant to your question but) tasks do accept a kwarg called wait_for which is an iterable list of either PrefectFuture or a task result which creates a state dependency

Thanks thats helpful