Issue 1:
Checklist

I have included the output of celery -A proj report in the issue.

(if you are not able to do this, then at least specify the Celery

version affected).

I have included all related issues and possible duplicate issues in this issue.

I have included the contents of pip freeze in the issue.

I have verified that the issue exists against the master branch of Celery.

I have tried reproducing the issue on more than one message broker and/or result backend.

I have tried reproducing the issue on more than one workers pool.

I have tried reproducing the issue with retries, ETA/Countdown & rate limits disabled.

Related Issues and Possible Duplicates

Related Issues

None

Possible Duplicates

None

Environment & Settings

Celery version: 4.2.1 (windowlicker) & 4.3.0rc1 (rhubarb)

celery report Output:

software -> celery:4.2.1 (windowlicker) kombu:4.2.1 py:3.6.4
billiard:3.5.0.3 redis:2.10.6
platform -> system:Darwin arch:64bit imp:CPython
loader -> celery.loaders.app.AppLoader
settings -> transport:redis results:django-db

and

software -> celery:4.3.0rc1 (rhubarb) kombu:4.3.0 py:3.6.4
billiard:3.6.0.0 py-amqp:2.4.1
platform -> system:Darwin arch:64bit
kernel version:18.2.0 imp:CPython
loader -> celery.loaders.app.AppLoader
settings -> transport:pyamqp results:disabled

Steps to Reproduce

Set up celery beat to run a task which has a custom queue configured (see example and the consequences below).

Required Dependencies

Minimal Python Version: 3.6.4

Minimal Broker Version: redis 4.0.8 / rabbitmq 3.7.5

Minimal Result Backend Version: Disabled / django 2.0.2

Minimal OS and/or Kernel Version: system:Darwin arch:64bit kernel version:18.2.0

Minimal Broker Client Version: redis:2.10.6 / py-amqp:2.4.1

Minimal Result Backend Client Version: Disabled / django-db

Python Packages

pip freeze Output:

amqp==2.4.1
billiard==3.5.0.5
celery==4.2.1
kombu==4.3.0
pytz==2018.9
vine==1.2.0

Other Dependencies

N/A

Minimally Reproducible Test Case

I couldn't figure a way to add the test to the integration tests; are there integration tests for celery beat?

Repository for the source of the following example: https://github.com/asfaltboy/celery_reproduce_5355

# tasks.py
from celery import Celery
from celery.signals import task_prerun

app = Celery('tasks', broker='pyamqp://guest@localhost//')

@app.task()
def add(x, y):
print(x + y)
return x + y

@app.task(queue='custom')
def multiply(x, y):
print(x * y)
return x * y

@task_prerun.connect()
def task_prerun(sender=None, task_id=None, task=None, **kwargs):
""" Log out some meta-info on the task being handled by worker """
print(
f"Handling {task.name} - {task_id} on queue: "
f"{sender.request.delivery_info['routing_key']}"
)

app.conf.beat_schedule = {
'add-every-2-seconds': {
'task': 'tasks.add',
'schedule': 2.0,
'args': (16, 16)
},
'multiple-every-2-seconds': {
'task': 'tasks.multiply',
'schedule': 2.0,
'args': (5, 5)
},
}
app.conf.timezone = 'UTC'

We launch the worker and beat (in sepearate windows):

$ pythonloc -m celery -A tasks worker -Q celery,custom

$ pythonloc -m celery -A tasks beat

And we prepare a script to schedule the tasks:

# run_delay.py
from tasks import add, multiply

print(add.delay(4, 4))
print(multiply.delay(2, 2))

The result of running add and multiply tasks by first calling delay() and then from beat:

[2019-02-22 22:37:19,666: WARNING/ForkPoolWorker-2] Handling tasks.add - d159d2cf-6135-49ba-ab81-f51ef0a52c47 on queue: celery
[2019-02-22 22:37:19,667: WARNING/ForkPoolWorker-2] 8
[2019-02-22 22:37:19,668: WARNING/ForkPoolWorker-1] Handling tasks.multiply - a1f09a7b-0a4c-4792-a35e-de705ed632f7 on queue: custom
[2019-02-22 22:37:19,669: WARNING/ForkPoolWorker-1] 4

[2019-02-22 22:37:21,955: WARNING/ForkPoolWorker-2] Handling tasks.add - ad8adc92-d190-4edc-9679-64e079e0076d on queue: celery
[2019-02-22 22:37:21,955: WARNING/ForkPoolWorker-2] 32
[2019-02-22 22:37:21,958: WARNING/ForkPoolWorker-1] Handling tasks.multiply - 476691f5-fe61-47d7-a7d3-431a63c530b6 on queue: custom
[2019-02-22 22:37:21,958: WARNING/ForkPoolWorker-1] 25

Expected Behavior

We expect both delay and beat to use the given task's queue option, as supplied to the task decorator (e.g custom for our multiply task).

Actual Behavior

Calling delay respects the given queue and runs the task in workers that consume from that queue. However, celery beat only scheduled the tasks on the default "celery" queue disregarding the queue= option of the task, thus not being consumed by workers consuming from the custom queue.

Issue 2:
Checklist

I have read the relevant section in the

contribution guide

on reporting bugs.

[X ] I have checked the issues list

for similar or identical bug reports.

[X ] I have checked the pull requests list

for existing proposed fixes.

[X ] I have checked the commit log

to find out if the bug was already fixed in the master branch.

[X ] I have included all related issues and possible duplicate issues

in this issue (If there are none, check this box anyway).

Mandatory Debugging Information

[X ] I have included the output of celery -A proj report in the issue.

(if you are not able to do this, then at least specify the Celery

version affected).

[X ] I have verified that the issue exists against the master branch of Celery.

I have included the contents of pip freeze in the issue.

[X ] I have included all the versions of all the external dependencies required

to reproduce this bug.

Optional Debugging Information

I have tried reproducing the issue on more than one Python version

and/or implementation.

I have tried reproducing the issue on more than one message broker and/or

result backend.

I have tried reproducing the issue on more than one version of the message

broker and/or result backend.

I have tried reproducing the issue on more than one operating system.

I have tried reproducing the issue on more than one workers pool.

I have tried reproducing the issue with autoscaling, retries,

ETA/Countdown & rate limits disabled.

I have tried reproducing the issue after downgrading

and/or upgrading Celery and its dependencies.

Related Issues and Possible Duplicates

Related Issues

None

Possible Duplicates

None

Environment & Settings

Celery version: 4.3.0

celery report Output:

-------------- celery@Parths-MacBook-Pro v4.3.0 (rhubarb)

---- **** -----

--- * *** * -- Darwin-18.2.0-x86_64-i386-64bit 2019-07-02 21:57:47

-- * - **** ---

** ---------- [config]

** ---------- .> app: default:0x10fe5bba8 (.default.Loader)

** ---------- .> transport: amqp://guest:**@localhost:5672//

** ---------- .> results: redis://localhost/

*** --- * --- .> concurrency: 8 (prefork)

-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)

--- ***** -----

-------------- [queues]

.> celery exchange=celery(direct) key=celery

Steps to Reproduce

Required Dependencies

Minimal Python Version: 3.6

Minimal Celery Version: 4.3.0

Minimal Kombu Version: N/A or Unknown

Minimal Broker Version: rabbitmq 3.7

Minimal Result Backend Version: Redis 3.2.1

Minimal OS and/or Kernel Version: N/A

Minimal Broker Client Version: N/A

Minimal Result Backend Client Version: N/A

Python Packages

pip freeze Output:

Other Dependencies

from celery import current_app

import os

import json

RABBITMQ_URL = os.getenv('RABBITMQ_URL', 'amqp://localhost')

REDIS_URL = os.getenv('REDIS_URL', 'redis://localhost')

CELERY_GLOBAL_CONFIG = {

'accept_content': ['json'],

'enable_utc': True,

}

CELERY_BROKER_CONFIG = {

'broker_url': RABBITMQ_URL,

'task_queue_ha_policy': 'all', # replicate queue to all broker nodes

}

CELERY_RESULT_BACKEND_CONFIG = {

'result_backend': REDIS_URL,

'result_compression': 'gzip',

'result_serializer': 'json',

'result_chord_join_timeout': 60*120, # 2hrs

}

CELERY_TASK_CONFIG = {

'task_serializer': 'json',

'task_time_limit': 60*240, # 4hrs

'task_routes': json.load(open('src/task_routes.json', 'rb')),

}

current_app.conf.update(CELERY_GLOBAL_CONFIG)

current_app.conf.update(CELERY_BROKER_CONFIG)

current_app.conf.update(CELERY_RESULT_BACKEND_CONFIG)

current_app.conf.update(CELERY_TASK_CONFIG)

Minimally Reproducible Test Case

@app.task
def consumer(doc_list, **kwargs):
chord([process_document.s(doc) for doc in doc_list])(aggregate.s())

@app.task
def process_document(doc):
return output

@app.task
def aggregate(document_outputs):
request.post(url, document_outputs, timeout=120)

Expected Behavior

I expect the aggregate callback to be called consistently after processing all the documents.

Actual Behavior

The aggregate callback is ONLY called when there are 1 or 2 documents, but not consistently called for any number of documents, even though headers are completing successfully.