Prefetch not working properly

[somehowchris]

Describe the bug

As I understand it. prefetch set to 1 and acks_late set to true should take on a task, do it, acknowledge and start again.

As of my case. prefetch doesn't limit the tasks consumed. It just takes on what ever it can take

To Reproduce
Steps to reproduce the behavior:

Use the prefetch and acks_late option on the consumer.

i.e.

let my_app = celery::app!(
	broker = AMQPBroker { &environment.amqp_addr },
	tasks = [training_tasks::train_environments],
	task_routes = [],
	acks_late=true,
	prefetch_count=1,
)
.await
.expect("Could not connect to celery rabbitmq broker");

#[cfg(debug_assertions)]
my_app.display_pretty().await;

my_app
    .consume()
    .await
    .expect("Failed to consume from broker");

Expected behavior

• Take task
• Process it
• Acknowlage
• Take on new one

Screenshots
If applicable, add screenshots to help explain your problem.

System (please complete the following information):

• OS: MacOS Monterey (12.2.1)
• Broker: AMQP
• Version:

celery = "0.4.0-rcn.11"