Kombu: Redeliver same tasks repeatedly with redis broker and gevent

[wuqiallen]

Hi,

kombu==4.6.4
celery==4.3.0
redis==3.2.1

The visibility_timeout is ignored when using redis broker and gevent. As the result, when launching multiple workers with redis broker and gevent, the tasks will be re-delivered repeatedly.
Maybe the reason is https://github.com/celery/kombu/pull/905.

def restore_visible(self, start=0, num=10, interval=10):
......
ceil = time() - self.visibility_timeout
......
env = _detect_environment()
if env == 'gevent':
    ceil = time()
visible = client.zrevrangebyscore(
    self.unacked_index_key, ceil, 0,
    start=num and start, num=num, withscores=True)
for tag, score in visible or []:
    self.restore_by_tag(tag, client)

When env==gevent, _ceil_ will be changed to _time()_ from _time() - self.visibility_timeout_.
As a result, all the tasks even the newly added ones in unacked set will be fetched out into _visible_, and then re-delivered by calling _restore_by_tag_, ignoring the functionality of visibility_timeout. The function _restore_visible_ in QoS is called by _maybe_restore_messages_

    def maybe_restore_messages(self):
        for channel in self._channels:
            if channel.active_queues:
                # only need to do this once, as they are not local to channel.
                return channel.qos.restore_visible(
                    num=channel.unacked_restore_limit,
                )

But the function _maybe_restore_messages_ is further called by other methods multiple times. Particularly in _register_with_event_loop_, where _maybe_restore_messages_ is called every 10 seconds.

    def register_with_event_loop(self, connection, loop):
        ...
        loop.call_repeatedly(10, cycle.maybe_restore_messages)

So when launching multiple workers with redis broker and gevent will re-deliver all tasks in unacked set repeatedly.

[matteius]

@wuqiallen I will ask that you also try this with the master branch of kombu in place of 4.6.4 -- I did a bunch of work with Redis recently that was merged: https://github.com/celery/kombu/pull/1089/files

While your issue report may be different, we need to know if it is still an issue for you on the master branch as we are prepping for a soon release. If we know it is still an issue than a possible fix may be discovered.

[matteius]

Also try master celery branch with that please, or at least latest. Also for what its worth latest redis on pypi is redis==3.3.8

[wuqiallen]

@matteius Thanks for your reply. I will try to reproduce this issue with the env which you talked me. I will let you know the results as soon as I can.

[arikfr]

This still reproduces on latest versions, as the relevant code is still there:

https://github.com/celery/kombu/blob/b51d1d678e198a80d7e5fd95f32674c7d8e04a75/kombu/transport/redis.py#L197-L199

Whenever restore_visible is called in a gevent environment it will restore all unacked messages, regardless of their visibility timeout. While #905 tried to fix an issue with not restoring messages when worker starts, they broke this functionality in all other cases (restore_visible is called periodically and not only when worker starts).

In our case, we don't even use gevent based workers. But our API services do use gevent. When we use celery.control to monitor the status of Celery (we have an API to return queues status), it triggers restore_visible with this broken code. It took us a while to find this :cry:

[deepanjan-nsk]

Can confirm @arikfr 's statement that we are still seeing this problem on 3rd feb 2019 as the code in question is still present.

[arikfr]

@deepanjan-nsk we ended up patching this in our codebase (and later replacing Celery with RQ). If you want I can try and find the patch.

[deepanjan-nsk]

Thanks @arikfr, We ended up doing the same w.r.t. a patch in our codebase.

[calvin620707]

Also, we do a patch to remove ceil = time() to avoid duplicated tasks since we don't care too much regarding losing tasks after workers restart.

[auvipy]

celery==4.4.*?

[calvin620707]

I had reproduced this with celery==4.4.0 and kombu==4.6.7. Actually, it seems like a kombu's issue instead of celery.

[auvipy]

try celery==4.4.2 and report again please

[mayouzi]

replace gevent with eventlet

[mayouzi]

replace gevent with eventlet

it`s necessary to remind that make a HTTP request cautious when use eventlet + requests
which need to install pyopenssl with pip install pyOpenSSL

[wuqiallen]

fixed on #1259