Celery chain breaks if one of the tasks fail

Question

What is the point of Celery chain if the whole chain breaks if one of the tasks fail?!!

I have this Celery chain:

res = chain(workme.s ( y=1111 ), workme2.s( 2222 ), workme3.s( 3333 ),)() 

And I made workme2 fails with retries like this:

@celery.task(default_retry_delay=5, max_retries = 10, queue="sure") 
def workme2(x,y):
    # try:      
    try:
        print str(y)
        sleep(2)
        print str(x)
        ## adding any condition that makes the task fail
        if x!=None:
            raise Exception('Aproblem from your workme task')
        print 'This is my username: ' + str(x['user']) + \
               ' And Password: ' + str(x['pas'])        
        return "22xx"
    except Exception, exc:
        workme2.retry(args=[x,y], exc=exc,)

Tergiversate

#11508612 – Calcariferous

@BernhardVallant, Hi, I downloaded the latest a couple of days ago, does this mean this patch is not included?? – Tergiversate

If it's newer than 3.0.4 i guess it should be included... – Calcariferous

I downloaded the latest version from the master branch on github, do I have to do something else. I just have one question for you .. did you try it yourself?? If you did and worked for you, please tell me yes, and send me the download link from github – Tergiversate

Answer 1

That is the point.

Forming a chain means your subtasks have some kind of serial dependency: Each one only makes sense if the previous one has been performed. Without this, you would simply use queueing or use a group rather than a chain.

So if one subtask fails (and still fails after attempting all its retries), the chain fails.

I readily admit that the documentation (as of Celery 3.1.18) is far from explicit in this respect, but the name suggests this semantics: "A chain is only as strong as its weakest link."

Forgat

Good point :) .. and I agree with you that there are some blurry points in the documentation – Tergiversate

Answer 2

The chains are designed to be stopped when there's a failure in the task. and it's the recommended way to design the workflow.

But for people, who needs and has specific requirement to continue on error ( or to go against the design ), here's how we can do it.

def ignorant_chain(task_list):
    prev = None
    for each_task in reversed(task_list):
        if prev is None:
            prev = each_task
            continue

        prev = each_task.set(link=prev).set(link_error=prev)

    return prev

# Usage 
ignorant_chain([
    task_pass.si("param_1", "param_2"),
    task_fail.si("param_1", "param_2"),
    task_pass.si("param_1", "param_2"),
]).apply_async()

Indraft