A question about priority queues with Celery

[Mahesh935525]

Describe the bug

RabbitMQ version - 3.12.7
celery - 3.1.23
amqp<2.0,>=1.4.9

The below code is used for priority queue for job

task.apply_async(args=[self.id], queue=priority, routing_key=priority)is

Reproduction steps

Expected behavior

The priority queue to be execute the job.

Additional context

No response

[lukebakken]

This information is insufficient to prove there is a bug in RabbitMQ by demonstrating the issue. You can't expect us to take the time to set up a Celery environment, guess at the code to use, and guess at how to reproduce this issue.

At the bare minimum, you should be able to provide a docker compose environment, or even a shell script, that will start RabbitMQ and your Celery worker and demonstrate the issue.

Please recognize the value of all of the free software you use, and the free support that comes with it, and make the effort to help maintainers help you.

[Mahesh935525]

Using Celery packages defined queue as defined in settings.py file

CELERY_QUEUES = (
Queue('default', routing_key='default'),
Queue('priority', routing_key='priority'),
Queue('scheduledTasks', routing_key='scheduledTasks'),
)

Based on the condition first importing job is to go priority queue
queue and routing_key need not be the same - see settings.py.

route = 'default'
diffSeconds = abs((self.modified - self.created).total_seconds())
if self.repeat == JobRepeat.NOREPEAT or diffSeconds < 1:
route = 'priority'
task.apply_async(args=[self.id], queue=route, routing_key=route)

Its always goes to default queue its not go to priority queue - But the same works for RabbitMQ 3.8.14 version

[lukebakken]

Please carefully re-read my request.

[Mahesh935525]

Hi ,

sorry - we not using docker compose and manually installed RabbitMQ and consumer in Azure container instance -

Installed RabbitMQ in Azure instance using - bitnami (RabbitMQ 3.12.7) - (using yml file)

Below is our environment details

python==2.7
amqp<2.0,>=1.4.9
billiard<3.4,>=3.3.0.23
celery==3.1.23
dj-database-url==0.2.1
django-celery==3.0.11
Django==1.4.3

The consumer instance to run worker we are running below commands in Azure container instance

Default worker -
[ "newrelic-admin", "run-program", "python", "manage.py", "celery", "worker", "--queues=default", "--concurrency=10", "--loglevel=INFO" ]

Priority worker -
[ "newrelic-admin", "run-program", "python", "manage.py", "celery", "worker", "--queues=priority", "--concurrency=1", "--loglevel=INFO" ]

scheduler-worker -
[ "newrelic-admin", "run-program", "python", "manage.py", "celery", "worker", "--beat", "--queues=scheduledTasks", "--concurrency=1", "--loglevel=INFO" ]

Note : settings.py - here we are defined celery queue

CELERY_QUEUES = (
Queue('default', routing_key='default'),
Queue('priority', routing_key='priority'),
Queue('breakingnews', routing_key='breakingnews'),
Queue('scheduledTasks', routing_key='scheduledTasks'),
)
CELERY_DEFAULT_QUEUE = 'default'
CELERY_DEFAULT_EXCHANGE = 'default'
CELERY_DEFAULT_EXCHANGE_TYPE = 'direct'
CELERY_DEFAULT_ROUTING_KEY = 'default'

[michaelklishin]

And what exactly are we supposed to observe, and what is expected?

Please help others help you, we are not very good at mind reading, and our small team cannot afford to guess and spend hours blindly troubleshooting every question with very incomplete information.

We need specific steps to reproduce and specific observations, plus an explanation of what exactly you expect to happen.

[michaelklishin]

We cannot suggest anything with a problem definition that consists of a claim that "this feature is not working" and a screenshot.

Please put together an executable way to reproduce (could be a Docker Compose file or a Python script, for example).

Priority classic queues have a number of caveats when there are consumers online or when QoS prefetch is used. They are reasonably well documented.