<h1 dir="auto">Checklist</h1>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have included the output of <code class="notranslate">celery -A proj report</code> in the issue.<br>
(if you are not able to do this, then at least specify the Celery<br>
version affected).</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have included all related issues and possible duplicate issues in this issue.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have included the contents of <code class="notranslate">pip freeze</code> in the issue.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have verified that the issue exists against the <code class="notranslate">master</code> branch of Celery.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have tried reproducing the issue on more than one message broker and/or result backend.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have tried reproducing the issue on more than one workers pool.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have tried reproducing the issue with retries, ETA/Countdown &amp; rate limits disabled.</li>
</ul>
<h2 dir="auto">Related Issues and Possible Duplicates</h2>
<h4 dir="auto">Related Issues</h4>
<ul dir="auto">
<li>None</li>
</ul>
<h4 dir="auto">Possible Duplicates</h4>
<ul dir="auto">
<li>None</li>
</ul>
<h2 dir="auto">Environment &amp; Settings</h2>
<p dir="auto"><strong>Celery version</strong>: 4.2.1 (windowlicker) &amp; 4.3.0rc1 (rhubarb)</p>
<details>
<summary><b><code class="notranslate">celery report</code> Output:</b></summary>
<p dir="auto">
</p><div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="software -&gt; celery:4.2.1 (windowlicker) kombu:4.2.1 py:3.6.4
            billiard:3.5.0.3 redis:2.10.6
platform -&gt; system:Darwin arch:64bit imp:CPython
loader   -&gt; celery.loaders.app.AppLoader
settings -&gt; transport:redis results:django-db"><pre class="notranslate"><code class="notranslate">software -&gt; celery:4.2.1 (windowlicker) kombu:4.2.1 py:3.6.4
            billiard:3.5.0.3 redis:2.10.6
platform -&gt; system:Darwin arch:64bit imp:CPython
loader   -&gt; celery.loaders.app.AppLoader
settings -&gt; transport:redis results:django-db
</code></pre></div>
<p dir="auto">and</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="software -&gt; celery:4.3.0rc1 (rhubarb) kombu:4.3.0 py:3.6.4
            billiard:3.6.0.0 py-amqp:2.4.1
platform -&gt; system:Darwin arch:64bit
            kernel version:18.2.0 imp:CPython
loader   -&gt; celery.loaders.app.AppLoader
settings -&gt; transport:pyamqp results:disabled"><pre class="notranslate"><code class="notranslate">software -&gt; celery:4.3.0rc1 (rhubarb) kombu:4.3.0 py:3.6.4
            billiard:3.6.0.0 py-amqp:2.4.1
platform -&gt; system:Darwin arch:64bit
            kernel version:18.2.0 imp:CPython
loader   -&gt; celery.loaders.app.AppLoader
settings -&gt; transport:pyamqp results:disabled
</code></pre></div>
<p dir="auto"></p>
</details>
<h1 dir="auto">Steps to Reproduce</h1>
<p dir="auto">Set up celery beat to run a task which has a custom queue configured (see example and the consequences below).</p>
<h2 dir="auto">Required Dependencies</h2>
<ul dir="auto">
<li><strong>Minimal Python Version</strong>: 3.6.4</li>
<li><strong>Minimal Broker Version</strong>: redis 4.0.8 / rabbitmq 3.7.5</li>
<li><strong>Minimal Result Backend Version</strong>: Disabled / django 2.0.2</li>
<li><strong>Minimal OS and/or Kernel Version</strong>: system:Darwin arch:64bit kernel version:18.2.0</li>
<li><strong>Minimal Broker Client Version</strong>: redis:2.10.6 / py-amqp:2.4.1</li>
<li><strong>Minimal Result Backend Client Version</strong>: Disabled / django-db</li>
</ul>
<h3 dir="auto">Python Packages</h3>
<details>
<summary><b><code class="notranslate">pip freeze</code> Output:</b></summary>
<p dir="auto">
</p><div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="amqp==2.4.1
billiard==3.5.0.5
celery==4.2.1
kombu==4.3.0
pytz==2018.9
vine==1.2.0"><pre class="notranslate"><code class="notranslate">amqp==2.4.1
billiard==3.5.0.5
celery==4.2.1
kombu==4.3.0
pytz==2018.9
vine==1.2.0
</code></pre></div>
<p dir="auto"></p>
</details>
<h3 dir="auto">Other Dependencies</h3>
<details>
<p dir="auto">
N/A
</p>
</details>
<h2 dir="auto">Minimally Reproducible Test Case</h2>
<p dir="auto">I couldn't figure a way to add the test to the integration tests; are there integration tests for celery beat?</p>
<p dir="auto">Repository for the source of the following example: <a href="https://github.com/asfaltboy/celery_reproduce_5355">https://github.com/asfaltboy/celery_reproduce_5355</a></p>
<details>
<p dir="auto">
</p><div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="# 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):
    &quot;&quot;&quot; Log out some meta-info on the task being handled by worker &quot;&quot;&quot;
    print(
        f&quot;Handling {task.name} - {task_id} on queue: &quot;
        f&quot;{sender.request.delivery_info['routing_key']}&quot;
    )


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'"><pre class="notranslate"><span class="pl-c"># tasks.py</span>
<span class="pl-k">from</span> <span class="pl-s1">celery</span> <span class="pl-k">import</span> <span class="pl-v">Celery</span>
<span class="pl-k">from</span> <span class="pl-s1">celery</span>.<span class="pl-s1">signals</span> <span class="pl-k">import</span> <span class="pl-s1">task_prerun</span>

<span class="pl-s1">app</span> <span class="pl-c1">=</span> <span class="pl-v">Celery</span>(<span class="pl-s">'tasks'</span>, <span class="pl-s1">broker</span><span class="pl-c1">=</span><span class="pl-s">'pyamqp://guest@localhost//'</span>)


<span class="pl-en">@<span class="pl-s1">app</span>.<span class="pl-en">task</span>()</span>
<span class="pl-k">def</span> <span class="pl-en">add</span>(<span class="pl-s1">x</span>, <span class="pl-s1">y</span>):
    <span class="pl-en">print</span>(<span class="pl-s1">x</span> <span class="pl-c1">+</span> <span class="pl-s1">y</span>)
    <span class="pl-k">return</span> <span class="pl-s1">x</span> <span class="pl-c1">+</span> <span class="pl-s1">y</span>


<span class="pl-en">@<span class="pl-s1">app</span>.<span class="pl-en">task</span>(<span class="pl-s1">queue</span><span class="pl-c1">=</span><span class="pl-s">'custom'</span>)</span>
<span class="pl-k">def</span> <span class="pl-en">multiply</span>(<span class="pl-s1">x</span>, <span class="pl-s1">y</span>):
    <span class="pl-en">print</span>(<span class="pl-s1">x</span> <span class="pl-c1">*</span> <span class="pl-s1">y</span>)
    <span class="pl-k">return</span> <span class="pl-s1">x</span> <span class="pl-c1">*</span> <span class="pl-s1">y</span>


<span class="pl-en">@<span class="pl-s1">task_prerun</span>.<span class="pl-en">connect</span>()</span>
<span class="pl-k">def</span> <span class="pl-en">task_prerun</span>(<span class="pl-s1">sender</span><span class="pl-c1">=</span><span class="pl-c1">None</span>, <span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-c1">None</span>, <span class="pl-s1">task</span><span class="pl-c1">=</span><span class="pl-c1">None</span>, <span class="pl-c1">**</span><span class="pl-s1">kwargs</span>):
    <span class="pl-s">""" Log out some meta-info on the task being handled by worker """</span>
    <span class="pl-en">print</span>(
        <span class="pl-s">f"Handling <span class="pl-s1"><span class="pl-kos">{</span><span class="pl-s1">task</span>.<span class="pl-s1">name</span><span class="pl-kos">}</span></span> - <span class="pl-s1"><span class="pl-kos">{</span><span class="pl-s1">task_id</span><span class="pl-kos">}</span></span> on queue: "</span>
        <span class="pl-s">f"<span class="pl-s1"><span class="pl-kos">{</span><span class="pl-s1">sender</span>.<span class="pl-s1">request</span>.<span class="pl-s1">delivery_info</span>[<span class="pl-s">'routing_key'</span>]<span class="pl-kos">}</span></span>"</span>
    )


<span class="pl-s1">app</span>.<span class="pl-s1">conf</span>.<span class="pl-s1">beat_schedule</span> <span class="pl-c1">=</span> {
    <span class="pl-s">'add-every-2-seconds'</span>: {
        <span class="pl-s">'task'</span>: <span class="pl-s">'tasks.add'</span>,
        <span class="pl-s">'schedule'</span>: <span class="pl-c1">2.0</span>,
        <span class="pl-s">'args'</span>: (<span class="pl-c1">16</span>, <span class="pl-c1">16</span>)
    },
    <span class="pl-s">'multiple-every-2-seconds'</span>: {
        <span class="pl-s">'task'</span>: <span class="pl-s">'tasks.multiply'</span>,
        <span class="pl-s">'schedule'</span>: <span class="pl-c1">2.0</span>,
        <span class="pl-s">'args'</span>: (<span class="pl-c1">5</span>, <span class="pl-c1">5</span>)
    },
}
<span class="pl-s1">app</span>.<span class="pl-s1">conf</span>.<span class="pl-s1">timezone</span> <span class="pl-c1">=</span> <span class="pl-s">'UTC'</span></pre></div>
<p dir="auto">We launch the worker and beat (in sepearate windows):</p>
<div class="highlight highlight-source-shell notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="$ pythonloc -m celery -A tasks worker -Q celery,custom

$ pythonloc -m celery -A tasks beat"><pre class="notranslate">$ pythonloc -m celery -A tasks worker -Q celery,custom

$ pythonloc -m celery -A tasks beat</pre></div>
<p dir="auto">And we prepare a script to schedule the tasks:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="# run_delay.py
from tasks import add, multiply

print(add.delay(4, 4))
print(multiply.delay(2, 2))"><pre class="notranslate"><span class="pl-c"># run_delay.py</span>
<span class="pl-k">from</span> <span class="pl-s1">tasks</span> <span class="pl-k">import</span> <span class="pl-s1">add</span>, <span class="pl-s1">multiply</span>

<span class="pl-en">print</span>(<span class="pl-s1">add</span>.<span class="pl-en">delay</span>(<span class="pl-c1">4</span>, <span class="pl-c1">4</span>))
<span class="pl-en">print</span>(<span class="pl-s1">multiply</span>.<span class="pl-en">delay</span>(<span class="pl-c1">2</span>, <span class="pl-c1">2</span>))</pre></div>
<p dir="auto">The result of running <code class="notranslate">add</code> and <code class="notranslate">multiply</code> tasks by first calling <code class="notranslate">delay()</code> and then from beat:</p>
<div class="highlight highlight-source-shell notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="[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"><pre class="notranslate">[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</pre></div>
<p dir="auto"></p>
</details>
<h1 dir="auto">Expected Behavior</h1>
<p dir="auto">We expect both delay and beat to use the given task's queue option, as supplied to the task decorator (e.g <code class="notranslate">custom</code> for our multiply task).</p>
<h1 dir="auto">Actual Behavior</h1>
<p dir="auto">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 <code class="notranslate">queue=</code> option of the task, thus not being consumed by workers consuming from the <code class="notranslate">custom</code> queue.</p>
<h1 dir="auto">Checklist</h1>

<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I have read the relevant section in the<br>
<a href="http://docs.celeryproject.org/en/latest/contributing.html#other-bugs" rel="nofollow">contribution guide</a><br>
on reporting bugs.</li>
<li>[X ] I have checked the <a href="https://github.com/celery/celery/issues?q=is%3Aissue+label%3A%22Issue+Type%3A+Bug+Report%22+-label%3A%22Category%3A+Documentation%22">issues list</a><br>
for similar or identical bug reports.</li>
<li>[X ] I have checked the <a href="https://github.com/celery/celery/pulls?q=is%3Apr+label%3A%22PR+Type%3A+Bugfix%22+-label%3A%22Category%3A+Documentation%22">pull requests list</a><br>
for existing proposed fixes.</li>
<li>[X ] I have checked the <a href="https://github.com/celery/celery/commits/master">commit log</a><br>
to find out if the bug was already fixed in the master branch.</li>
<li>[X ] I have included all related issues and possible duplicate issues<br>
in this issue (If there are none, check this box anyway).</li>
</ul>
<h2 dir="auto">Mandatory Debugging Information</h2>
<ul class="contains-task-list">
<li>[X ] I have included the output of <code class="notranslate">celery -A proj report</code> in the issue.<br>
(if you are not able to do this, then at least specify the Celery<br>
version affected).</li>
<li>[X ] I have verified that the issue exists against the <code class="notranslate">master</code> branch of Celery.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> I have included the contents of <code class="notranslate">pip freeze</code> in the issue.</li>
<li>[X ] I have included all the versions of all the external dependencies required<br>
to reproduce this bug.</li>
</ul>
<h2 dir="auto">Optional Debugging Information</h2>

<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> I have tried reproducing the issue on more than one Python version<br>
and/or implementation.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> I have tried reproducing the issue on more than one message broker and/or<br>
result backend.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> I have tried reproducing the issue on more than one version of the message<br>
broker and/or result backend.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> I have tried reproducing the issue on more than one operating system.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> I have tried reproducing the issue on more than one workers pool.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> I have tried reproducing the issue with autoscaling, retries,<br>
ETA/Countdown &amp; rate limits disabled.</li>
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> I have tried reproducing the issue after downgrading<br>
and/or upgrading Celery and its dependencies.</li>
</ul>
<h2 dir="auto">Related Issues and Possible Duplicates</h2>

<h4 dir="auto">Related Issues</h4>
<ul dir="auto">
<li>None</li>
</ul>
<h4 dir="auto">Possible Duplicates</h4>
<ul dir="auto">
<li>None</li>
</ul>
<h2 dir="auto">Environment &amp; Settings</h2>

<p dir="auto"><strong>Celery version</strong>: 4.3.0</p>

<details>
<summary><b><code class="notranslate">celery report</code> Output:</b></summary>
<p dir="auto">
</p><p dir="auto">-------------- celery@Parths-MacBook-Pro v4.3.0 (rhubarb)<br>
---- **** -----<br>
--- * ***  * -- Darwin-18.2.0-x86_64-i386-64bit 2019-07-02 21:57:47<br>
-- * - **** ---</p>
<ul dir="auto">
<li>** ---------- [config]</li>
<li>** ---------- .&gt; app:         default:0x10fe5bba8 (.default.Loader)</li>
<li>** ---------- .&gt; transport:   amqp://guest:**<a class="user-mention notranslate" data-hovercard-type="user" data-hovercard-url="/users/localhost/hovercard" data-octo-click="hovercard-link-click" data-octo-dimensions="link_type:self" href="https://github.com/localhost">@localhost</a>:5672//</li>
<li>** ---------- .&gt; results:     redis://localhost/</li>
<li>*** --- * --- .&gt; concurrency: 8 (prefork)<br>
-- ******* ---- .&gt; task events: OFF (enable -E to monitor tasks in this worker)<br>
--- ***** -----<br>
-------------- [queues]<br>
.&gt; celery           exchange=celery(direct) key=celery</li>
</ul>
<p dir="auto"></p>
</details>
<h1 dir="auto">Steps to Reproduce</h1>
<h2 dir="auto">Required Dependencies</h2>

<ul dir="auto">
<li><strong>Minimal Python Version</strong>: 3.6</li>
<li><strong>Minimal Celery Version</strong>: 4.3.0</li>
<li><strong>Minimal Kombu Version</strong>: N/A or Unknown</li>
<li><strong>Minimal Broker Version</strong>: rabbitmq 3.7</li>
<li><strong>Minimal Result Backend Version</strong>: Redis 3.2.1</li>
<li><strong>Minimal OS and/or Kernel Version</strong>: N/A</li>
<li><strong>Minimal Broker Client Version</strong>: N/A</li>
<li><strong>Minimal Result Backend Client Version</strong>: N/A</li>
</ul>
<h3 dir="auto">Python Packages</h3>

<details>
<summary><b><code class="notranslate">pip freeze</code> Output:</b></summary>
<p dir="auto">
</p><div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content=""><pre class="notranslate"><code class="notranslate"></code></pre></div>
<p dir="auto"></p>
</details>
<h3 dir="auto">Other Dependencies</h3>

<details>
<p dir="auto">
from celery import current_app
</p><p dir="auto">import os<br>
import json</p>
<p dir="auto">RABBITMQ_URL = os.getenv('RABBITMQ_URL', 'amqp://localhost')<br>
REDIS_URL = os.getenv('REDIS_URL', 'redis://localhost')</p>
<p dir="auto">CELERY_GLOBAL_CONFIG = {<br>
'accept_content': ['json'],<br>
'enable_utc': True,<br>
}</p>
<p dir="auto">CELERY_BROKER_CONFIG = {<br>
'broker_url': RABBITMQ_URL,<br>
'task_queue_ha_policy': 'all', # replicate queue to all broker nodes<br>
}</p>
<p dir="auto">CELERY_RESULT_BACKEND_CONFIG = {<br>
'result_backend': REDIS_URL,<br>
'result_compression': 'gzip',<br>
'result_serializer': 'json',<br>
'result_chord_join_timeout': 60*120, # 2hrs<br>
}</p>
<p dir="auto">CELERY_TASK_CONFIG = {<br>
'task_serializer': 'json',<br>
'task_time_limit': 60*240, # 4hrs<br>
'task_routes': json.load(open('src/task_routes.json', 'rb')),<br>
}</p>
<p dir="auto">current_app.conf.update(CELERY_GLOBAL_CONFIG)<br>
current_app.conf.update(CELERY_BROKER_CONFIG)<br>
current_app.conf.update(CELERY_RESULT_BACKEND_CONFIG)<br>
current_app.conf.update(CELERY_TASK_CONFIG)</p>
<p dir="auto"></p>
</details>
<h2 dir="auto">Minimally Reproducible Test Case</h2>

<details>
<p dir="auto">
</p><div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="@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)"><pre class="notranslate"><span class="pl-en">@<span class="pl-s1">app</span>.<span class="pl-s1">task</span></span>
<span class="pl-k">def</span> <span class="pl-en">consumer</span>(<span class="pl-s1">doc_list</span>, <span class="pl-c1">**</span><span class="pl-s1">kwargs</span>):
    <span class="pl-en">chord</span>([<span class="pl-s1">process_document</span>.<span class="pl-en">s</span>(<span class="pl-s1">doc</span>) <span class="pl-k">for</span> <span class="pl-s1">doc</span> <span class="pl-c1">in</span> <span class="pl-s1">doc_list</span>])(<span class="pl-s1">aggregate</span>.<span class="pl-en">s</span>())

<span class="pl-en">@<span class="pl-s1">app</span>.<span class="pl-s1">task</span></span>
<span class="pl-k">def</span> <span class="pl-en">process_document</span>(<span class="pl-s1">doc</span>):
    <span class="pl-k">return</span> <span class="pl-s1">output</span>

<span class="pl-en">@<span class="pl-s1">app</span>.<span class="pl-s1">task</span></span>
<span class="pl-k">def</span> <span class="pl-en">aggregate</span>(<span class="pl-s1">document_outputs</span>):
    <span class="pl-s1">request</span>.<span class="pl-en">post</span>(<span class="pl-s1">url</span>, <span class="pl-s1">document_outputs</span>, <span class="pl-s1">timeout</span><span class="pl-c1">=</span><span class="pl-c1">120</span>)</pre></div>
<p dir="auto"></p>
</details>
<h1 dir="auto">Expected Behavior</h1>

<p dir="auto">I expect the aggregate callback to be called consistently after processing all the documents.</p>
<h1 dir="auto">Actual Behavior</h1>

<p dir="auto">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.</p>