Repository: celery/celery
File: celery/worker/strategy.py (Python)

Code before:
    # (optimized to avoid calling request.send_event)
    eventer = consumer.event_dispatcher
    events = eventer and eventer.enabled
    send_event = eventer and eventer.send
    task_sends_events = events and task.send_events

    call_at = consumer.timer.call_at
    apply_eta_task = consumer.apply_eta_task
    rate_limits_enabled = not consumer.disable_rate_limits
    get_bucket = consumer.task_buckets.__getitem__
    handle = consumer.on_task_request
    limit_task = consumer._limit_task
    limit_post_eta = consumer._limit_post_eta
    Request = symbol_by_name(task.Request)
    Req = create_request_cls(Request, task, consumer.pool, hostname, eventer,
                             app=app)

    revoked_tasks = consumer.controller.state.revoked
    eta_task_limit = getattr(app.conf, 'worker_eta_task_limit', None)

    # Apply ETA task limit to QoS prefetch count if configured
    if eta_task_limit is not None:
        # Set a maximum prefetch count to limit ETA tasks
        original_prefetch = consumer.qos.value
        consumer.qos.value = min(original_prefetch, eta_task_limit)
        consumer.qos.set(consumer.qos.value)

    def task_message_handler(message, body, ack, reject, callbacks,
                             to_timestamp=to_timestamp):
        if body is None and 'args' not in message.payload:
            body, headers, decoded, utc = (
                message.body, message.headers, False, app.uses_utc_timezone(),
            )
        else:
            if 'args' in message.payload:
                body, headers, decoded, utc = hybrid_to_proto2(message,
                                                               message.payload)
            else:
                body, headers, decoded, utc = proto1_to_proto2(message, body)

        req = Req(
            message,
            on_ack=ack, on_reject=reject, app=app, hostname=hostname,
            eventer=eventer, task=task, connection_errors=connection_errors,
            body=body, headers=headers, decoded=decoded, utc=utc,
        )
        if _does_info:
            # Similar to `app.trace.info()`, we pass the formatting args as the
            # `extra` kwarg for custom log handlers
            context = {
                'id': req.id,

Review comment (refactor):
Thanks for the quick update ! But are you sure this works ? It seems that this code would be executed only once during the strategy initialization ? But the actual incrementation of prefetch count is done on each task fetching inside `task_message_handler` ([here](https://github.com/celery/celery/blob/v5.5.3/celery/worker/strategy.py#L198) for example). So maybe it should be moved there.

However, I still believe that this should more appear in kombu ? For example when you define the consumer or consumer.qos you would pass it the prefetch limit. And when you call `consumer.qos.increment_eventually()` it would not increment after the limit is reached. The benefit of this approach is that you don't violate the single responsibility principle, and the cap of prefetch count would be transparent here when done in kombu. Basically, you won't need any change here in strategy apart from passing that prefetch limit when you initialize the consumer or consumer.qos whichever is initialized here. 

Code after:
    # (optimized to avoid calling request.send_event)
    eventer = consumer.event_dispatcher
    events = eventer and eventer.enabled
    send_event = eventer and eventer.send
    task_sends_events = events and task.send_events

    call_at = consumer.timer.call_at
    apply_eta_task = consumer.apply_eta_task
    rate_limits_enabled = not consumer.disable_rate_limits
    get_bucket = consumer.task_buckets.__getitem__
    handle = consumer.on_task_request
    limit_task = consumer._limit_task
    limit_post_eta = consumer._limit_post_eta
    Request = symbol_by_name(task.Request)
    Req = create_request_cls(Request, task, consumer.pool, hostname, eventer,
                             app=app)

    revoked_tasks = consumer.controller.state.revoked

    def task_message_handler(message, body, ack, reject, callbacks,
                             to_timestamp=to_timestamp):
        if body is None and 'args' not in message.payload:
            body, headers, decoded, utc = (
                message.body, message.headers, False, app.uses_utc_timezone(),
            )
        else:
            if 'args' in message.payload:
                body, headers, decoded, utc = hybrid_to_proto2(message,
                                                               message.payload)
            else:
                body, headers, decoded, utc = proto1_to_proto2(message, body)

        req = Req(
            message,
            on_ack=ack, on_reject=reject, app=app, hostname=hostname,
            eventer=eventer, task=task, connection_errors=connection_errors,
            body=body, headers=headers, decoded=decoded, utc=utc,
        )
        if _does_info:
            # Similar to `app.trace.info()`, we pass the formatting args as the
            # `extra` kwarg for custom log handlers
            context = {
                'id': req.id,
                'name': req.name,
                'args': req.argsrepr,
                'kwargs': req.kwargsrepr,
                'eta': req.eta,
            }
            info(_app_trace.LOG_RECEIVED, context, extra={'data': context})
        if (req.expires or req.id in revoked_tasks) and req.revoked():
            return