Celery qpid revoke bug

revoketest.py

"""
Reproducer for celery task revoke bugs.

Save this file as something like "revoketest.py" and run it to execute
testcases.

Includes three subcommands:

    worker:
        Use this to run the celery worker

    revoke1:
        Test case: revoke a task which is in progress, with terminate=True,
        while there are no other tasks in the queue. Then submit more tasks.

        Expected behavior: the task is terminated; subsequent tasks run:
            task1: 4
            Sent task2 revoke
            task2: terminated
            task3: 2

    revoke2:
        Test case: revoke a task which is in progress, with terminate=True,
        while there are other tasks in the queue. Then submit more tasks.

        Expected behavior: the task is terminated; subsequent tasks run
            task1: 4
            Sent task2 revoke
            task2: terminated
            task3: 4
            task4: 2


Broker can be selected by BROKER environment variable. Observed behavior is:

    BROKER=    (rabbitmq, the default)
        Both tests pass

    BROKER=redis://
        Both tests pass

    BROKER=qpid://
        In revoke1, the task is terminated, but worker then hangs and does
        not process any more tasks:
            task1: 4
            Sent task2 revoke
            task2: terminated
            # BUG: crashes due to timeout for task3

        In revoke2, the task is not terminated. The worker completes the
        revoked task and continues to process subsequent tasks.
            task1: 4
            Sent task2 revoke
            task2: 5   # BUG: why wasn't it terminated?
            task3: 4
            task4: 2

Tested with:

    amqp==2.4.2
    billiard==3.5.0.5
    celery==4.2.2
    kombu==4.3.0
    pytz==2018.9
    qpid-python==1.36.0.post1
    qpid-tools==1.36.0.post1
    redis==3.2.1
    vine==1.3.0
"""

from __future__ import print_function
import argparse
import time
import os
import sys

from celery import Celery

TIMEOUT = 30

module_name = os.path.basename(__file__)
if module_name.endswith('.py'):
    module_name = module_name[:-3]

app = Celery(module_name, backend='rpc://',
    broker=os.environ.get('BROKER'))


@app.task
def do_something(name, delay):
    print("%s: now waiting %s" % (name, delay))
    time.sleep(delay)
    print("%s: waited %s" % (name, delay))
    return delay


def run_worker():
    basename = os.path.basename(__file__)
    basename = basename[:-3]
    os.execv(sys.executable, [
        sys.executable,
        '-mcelery',
        'worker',
        # Point at this module
        '-A', basename,
        '--loglevel=info',
        '-c1',
    ])


def run_revoke1():
    """Revoke testcase 1: revoke & terminate a task while the worker
    has no other tasks in queue, then try to run another task
    """
    task1 = do_something.delay('task1', 4)
    task2 = do_something.delay('task2', 5)

    print("task1: %s" % task1.get(TIMEOUT))

    # give some time for worker to start task2
    time.sleep(1.0)
    task2.revoke(terminate=True)
    print("Sent task2 revoke")


    try:
        print("task2: %s" % task2.get(TIMEOUT))
    except Exception as e:
        print("task2: %s" % e)

    task3 = do_something.delay('task3', 2)
    print("task3: %s" % task3.get(TIMEOUT))


def run_revoke2():
    """Revoke testcase 2: revoke & terminate a task while the worker
    has later tasks in queue, then try to run another task
    """
    task1 = do_something.delay('task1', 4)
    task2 = do_something.delay('task2', 5)
    task3 = do_something.delay('task3', 4)

    print("task1: %s" % task1.get(TIMEOUT))

    # give some time for worker to start task2
    time.sleep(1.0)
    task2.revoke(terminate=True)
    print("Sent task2 revoke")

    try:
        # Expected behavior: task2 should be terminated
        # Actual behavior: task2 completes successfully here
        print("task2: %s" % task2.get(TIMEOUT))
    except Exception as e:
        print("task2: %s" % e)

    task4 = do_something.delay('task4', 2)

    print("task3: %s" % task3.get(TIMEOUT))
    print("task4: %s" % task4.get(TIMEOUT))


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('mode',
               choices=['worker', 'revoke1', 'revoke2'])
    p = parser.parse_args()

    if p.mode == 'worker':
        run_worker()
    elif p.mode == 'revoke1':
        run_revoke1()
    elif p.mode == 'revoke2':
        run_revoke2()


if __name__ == '__main__':
    main()