Hello, can someone help me clear this doubt: While using Prefect 2, if a flow fails then the retry option would run the entire flow again or it would resume the flow from the task or sub-flow where it failed? I had tried this previously and it restarted the entire flow, curious to know if this is made available in any new version. This particular feature works in Prefect 1 though.

Hi @Sagar Dhande, for more granular retries (on tasks), you can either set your prefect

PREFECT_TASK_DEFAULT_RETRIES

setting to non-zero to automatically retry failed tasks or else pass it in to the specific task in question as an argument like

@task(retries=10)

. Here is a full working example:

from prefect import flow, task
from prefect.logging import get_run_logger
@task
def always_succeed():
    # task that should run once
    logger = get_run_logger()
    <http://logger.info|logger.info>("This is fine.")
    
@task(retries=10)  # by configuring retries on a task level, it means that we won't have to run the entire flow again if this task fails
def often_fail():
    # task that should run many times
    import random
    if random.random() < 0.9:
        raise ValueError("This should happen often")
    return "This is fine."

@flow(retries=2)
def sometimes_failing_flow():
    always_succeed()
    often_fail()

if __name__ == "__main__":
    sometimes_failing_flow()

Even if you remove flow-level retries from this example, the task itself will be retried up to 10 times. Adding

retries=2

on the flow here means that if the 10 task retries still leads to a failed state, it will then try to run the flow again (and I think it would still skip the

always_succeed()

task).

Thanks a lot @Serina thats what I needed