t/integration/tasks.py

@@ -178,3 +178,13 @@ def build_chain_inside_task(self):
    )
    result = test_chain()
    return result


@shared_task
def fail(*args):
    raise Exception('Task expected to fail')


@shared_task
def chord_error(args):
    return args

t/integration/test_canvas.py

@@ -12,7 +12,8 @@
from .tasks import (add, add_chord_to_chord, add_replaced, add_to_all,
                    add_to_all_to_chord, build_chain_inside_task, collect_ids,
                    delayed_sum, delayed_sum_with_soft_guard, identity, ids,
                    print_unicode, redis_echo, second_order_replace1, tsum)
                    print_unicode, redis_echo, second_order_replace1, tsum,
                    fail, chord_error)

TIMEOUT = 120

@@ -521,3 +522,38 @@ def assert_parentids_chord(self, res, expected_root_id):
        assert value == 1
        assert root_id == expected_root_id
        assert parent_id is None

    def test_chord_on_error(self, manager):
        from celery import states

        # Run the chord and wait for the error callback to finish.
        c1 = chord(
            header=[add.s(1, 2), add.s(3, 4), fail.s()],
            body=print_unicode.s('This should not be called').on_error(
                chord_error.s()),
        )
        res = c1()
        res.get(propagate=False)
        res.children[0].children[0].get(propagate=False)

        # Use the error callback's result to find the failed task.
        error_callback_result = AsyncResult(res.children[0].children[0].result)
        failed_task_id = error_callback_result.result.args[0].split()[3]

        # Use new group_id result metadata to get group ID.
        failed_task_result = AsyncResult(failed_task_id)
        original_group_id = failed_task_result._get_task_meta()['group_id']

        # Use group ID to get preserved group result.
        backend = fail.app.backend
        j_key = backend.get_key_for_group(original_group_id, '.j')
        redis_connection = get_redis_connection()
        chord_results = [backend.decode(t) for t in
                         redis_connection.lrange(j_key, 0, 3)]

        # Validate group result
        assert [cr[3] for cr in chord_results if cr[2] == states.SUCCESS] == \
               [3, 7]

        assert len([cr for cr in chord_results if cr[2] != states.SUCCESS]
                   ) == 1