<p dir="auto">We're using Celery with MongoDB both as broker and results backend in production with quite a success. However recently we found that Mongo database used for Celery queues grown up to 50+ GBs. After short investigation I found that 99.9% of all documents in "messages" collection are belonging to "celeryev" queue.</p>
<p dir="auto">Our workers have events enabled in order for Celery Flower frontend to work. I always assumed that event got wiped after consuming, but looks like I was wrong.</p>
<p dir="auto">Celery report is as follows:</p>
<p dir="auto">[root@app1 nimble]# bin/celery report</p>
<p dir="auto">software -&gt; celery:3.0.9nimble (Chiastic Slide) kombu:2.4.7nimble py:2.7.3<br>
billiard:2.7.3.12 pymongo:2.1.1<br>
platform -&gt; system:Linux arch:64bit, ELF imp:CPython<br>
loader   -&gt; celery.loaders.default.Loader<br>
settings -&gt; transport:mongodb results:mongodb</p>
<p dir="auto">I believe that we see exactly the same behavior as in <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="1251771" data-permission-text="Title is private" data-url="https://github.com/celery/celery/issues/436" data-hovercard-type="issue" data-hovercard-url="/celery/celery/issues/436/hovercard" href="https://github.com/celery/celery/issues/436">#436</a>. I see three different celeryev queues, but we only have one single listener active (Flower):</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="SECONDARY&gt; db.messages.distinct(&quot;queue&quot;, {queue: /^celeryev/})
[
    &quot;celeryev.2a304cc3-78b8-451c-9728-99d0c09ccde1&quot;,
    &quot;celeryev.3cbbd9b5-b967-4297-8a50-f3313dae0650&quot;,
    &quot;celeryev.d94fb4c0-8ab4-4d12-9469-f93e090a7631&quot;
]"><pre class="notranslate"><code class="notranslate">SECONDARY&gt; db.messages.distinct("queue", {queue: /^celeryev/})
[
    "celeryev.2a304cc3-78b8-451c-9728-99d0c09ccde1",
    "celeryev.3cbbd9b5-b967-4297-8a50-f3313dae0650",
    "celeryev.d94fb4c0-8ab4-4d12-9469-f93e090a7631"
]
</code></pre></div>
<p dir="auto">I also see a lot of "worker.heartbeat" routing keys in documents in these queues. Probably that's the majority of all events.</p>
<p dir="auto">Distribution between these three queues is quite uniform:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="SECONDARY&gt; db.messages.group({key: {queue: true}, cond: {queue: /^celeryev/}, reduce: function(obj,prev) { prev.csum += 1; }, initial: { csum: 0 }})
[
    {
        &quot;queue&quot; : &quot;celeryev.2a304cc3-78b8-451c-9728-99d0c09ccde1&quot;,
        &quot;csum&quot; : 21619432
    },
    {
        &quot;queue&quot; : &quot;celeryev.3cbbd9b5-b967-4297-8a50-f3313dae0650&quot;,
        &quot;csum&quot; : 14117210
    },
    {
        &quot;queue&quot; : &quot;celeryev.d94fb4c0-8ab4-4d12-9469-f93e090a7631&quot;,
        &quot;csum&quot; : 18428199
    }
]"><pre class="notranslate"><code class="notranslate">SECONDARY&gt; db.messages.group({key: {queue: true}, cond: {queue: /^celeryev/}, reduce: function(obj,prev) { prev.csum += 1; }, initial: { csum: 0 }})
[
    {
        "queue" : "celeryev.2a304cc3-78b8-451c-9728-99d0c09ccde1",
        "csum" : 21619432
    },
    {
        "queue" : "celeryev.3cbbd9b5-b967-4297-8a50-f3313dae0650",
        "csum" : 14117210
    },
    {
        "queue" : "celeryev.d94fb4c0-8ab4-4d12-9469-f93e090a7631",
        "csum" : 18428199
    }
]
</code></pre></div>
<p dir="auto">I'm wondering, is it possible to use same fix for MongoDB as was used for Redis in <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="1251771" data-permission-text="Title is private" data-url="https://github.com/celery/celery/issues/436" data-hovercard-type="issue" data-hovercard-url="/celery/celery/issues/436/hovercard" href="https://github.com/celery/celery/issues/436">#436</a>?</p>
<p dir="auto">I am using Django-celery and Redis backend to push the tasks in celery queue. I am pushing tasks with same custom task_id in the queue. The documentation says that task_id is UUID which should be unique. But I am able to add more than one task with the same task_id in the queue. It is expected that it should raise an exception if a task with the same task_id is already there in the queue.</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="@task
def register_task(a, b):
    print a + b

@task
def deregister_task(a, b, c):
    print a + b + c

taskid=str(UUID(md5('test').hexdigest()))
register_task.apply_async(args=('hello', 'world'), task_id=taskid, countdown=60)
deregister_task.apply_async(args=('sample', 'hello', 'world'), task_id=taskid, countdown=60)
i = inspect()
i.scheduled()

{u'celery@Suyashs-MacBook-Pro.local': [{u'eta': u'2017-06-05T17:15:33.947002',
   u'priority': 6,
   u'request': {u'acknowledged': False,
    u'args': u&quot;('sample', 'hello', 'world')&quot;,
    u'delivery_info': {u'exchange': u'celery',
     u'priority': 0,
     u'redelivered': None,
     u'routing_key': u'celery'},
    u'hostname': u'celery@Suyashs-MacBook-Pro.local',
    u'id': u'098f6bcd-4621-d373-cade-4e832627b4f6',
    u'kwargs': u'{}',
    u'name': u'flight.tasks.deregister_task',
    u'time_start': None,
    u'worker_pid': None}},
  {u'eta': u'2017-06-05T17:15:44.234537',
   u'priority': 6,
   u'request': {u'acknowledged': False,
    u'args': u&quot;('hello', 'world')&quot;,
    u'delivery_info': {u'exchange': u'celery',
     u'priority': 0,
     u'redelivered': None,
     u'routing_key': u'celery'},
    u'hostname': u'celery@Suyashs-MacBook-Pro.local',
    u'id': u'098f6bcd-4621-d373-cade-4e832627b4f6',
    u'kwargs': u'{}',
    u'name': u'flight.tasks.register_task',
    u'time_start': None,
    u'worker_pid': None}}]}

"><pre class="notranslate"><code class="notranslate">@task
def register_task(a, b):
    print a + b

@task
def deregister_task(a, b, c):
    print a + b + c

taskid=str(UUID(md5('test').hexdigest()))
register_task.apply_async(args=('hello', 'world'), task_id=taskid, countdown=60)
deregister_task.apply_async(args=('sample', 'hello', 'world'), task_id=taskid, countdown=60)
i = inspect()
i.scheduled()

{u'celery@Suyashs-MacBook-Pro.local': [{u'eta': u'2017-06-05T17:15:33.947002',
   u'priority': 6,
   u'request': {u'acknowledged': False,
    u'args': u"('sample', 'hello', 'world')",
    u'delivery_info': {u'exchange': u'celery',
     u'priority': 0,
     u'redelivered': None,
     u'routing_key': u'celery'},
    u'hostname': u'celery@Suyashs-MacBook-Pro.local',
    u'id': u'098f6bcd-4621-d373-cade-4e832627b4f6',
    u'kwargs': u'{}',
    u'name': u'flight.tasks.deregister_task',
    u'time_start': None,
    u'worker_pid': None}},
  {u'eta': u'2017-06-05T17:15:44.234537',
   u'priority': 6,
   u'request': {u'acknowledged': False,
    u'args': u"('hello', 'world')",
    u'delivery_info': {u'exchange': u'celery',
     u'priority': 0,
     u'redelivered': None,
     u'routing_key': u'celery'},
    u'hostname': u'celery@Suyashs-MacBook-Pro.local',
    u'id': u'098f6bcd-4621-d373-cade-4e832627b4f6',
    u'kwargs': u'{}',
    u'name': u'flight.tasks.register_task',
    u'time_start': None,
    u'worker_pid': None}}]}


</code></pre></div>