Continuously receiving unsubscribe message causing CPU 100% in drain_events() loop

[viennadd]

Hi all,

the following code snippet works fine when using kombu 3.x,
when I have upgraded to kombu 4.6.7, it continuously receiving an unsubscribe message to our exchange

since it always can receive messages (unsubscribe messages), the drain_events() will always return in a very short time, it then causes CPU 100%, anyone got a clue?

Code snippet to reproduce:

import logging
import time
import uuid
from threading import Thread

from kombu import Connection, connections, Exchange, Queue

broker_url = "redis://:****/1"


def subscribe_messages(connection, message_queue):
    def message_callback(body, message):
        print(body)
        message.ack()

    conn = connections[connection].acquire(block=True)
    while True:
        try:
            with conn.Consumer(queues=message_queue, callbacks=[message_callback]):
                conn.drain_events(timeout=10)
                print("drained")
        except Exception as e:
            logging.exception(e)


def main():
    connection = Connection(broker_url)
    exchange = Exchange(name="example_exchange", type="fanout")
    queue = Queue(name=str(uuid.uuid4()), exchange=exchange)

    thread = Thread(name="subscribe_thread", target=subscribe_messages, daemon=True, kwargs={
        "connection": connection,
        "message_queue": queue,
    }).start()

    producer = connection.Producer(exchange=exchange)
    producer.publish(body={"msg": "aaaa", "channels": "bbbb"})

    time.sleep(1000)


if __name__ == '__main__':
    main()

pip freeze: (redis==3.4.1, kombu==4.6.7)

amqp==2.5.2
aniso8601==7.0.0
anyjson==0.3.3
attrs==19.1.0
beautifulsoup4==4.8.2
bidict==0.17.5
billiard==3.6.3.0
blinker==1.4
cachetools==2.0.1
celery==4.4.0
certifi==2019.6.16
chardet==3.0.4
cheroot==6.0.0
click==6.7
colorama==0.3.9
ConcurrentLogHandler==0.9.1
cxxfilt==0.2.0
decorator==4.4.0
django-indexer==0.3.0
et-xmlfile==1.0.1
flasgger==0.9.0
Flask==0.12.2
Flask-Caching==1.7.1
Flask-Compress==1.4.0
Flask-Cors==3.0.3
Flask-Login==0.4.1
Flask-RESTful==0.3.6
Flask-WTF==0.14.2
futures==3.0.5
fuzzyfinder==1.0.0
gitdb2==2.0.5
GitPython==2.1.11
gunicorn==20.0.4
idna==2.6
importlib-metadata==0.19
isort==4.2.5
itsdangerous==1.1.0
jdcal==1.4.1
jenkinsapi==0.3.6
Jinja2==2.10.1
jsonschema==3.0.2
kombu==4.6.7
lazy-object-proxy==1.4.2
ldap3==2.6
lxml==4.1.1
Markdown==3.1.1
MarkupSafe==1.1.1
minio==3.0.1
mistune==0.8.4
more-itertools==7.2.0
networkx==1.11
openpyxl==2.4.9
pathtools==0.1.2
pendulum==2.0.5
phply==1.0.0
pluggy==0.12.0
ply==3.11
prettytable==0.7.2
prompt-toolkit==1.0.15
protobuf==3.6.0
psutil==5.4.3
pyasn1==0.4.6
pycryptodome==3.4.7
pyelftools==0.25
PyJWT==1.7.1
pymongo==3.9.0
PyPrind==2.11.1
pyrsistent==0.15.4
python-dateutil==2.8.1
python-gitlab==1.6.0
pytz==2019.2
pytzdata==2019.3
PyYAML==5.1
raven==6.6.0
rdbtools==0.1.12
redis==3.4.1
requests==2.18.4
sentry-sdk==0.9.2
setuptools-scm==3.5.0
simplejson==3.10.0
six==1.12.0
smmap2==2.0.5
soupsieve==1.9.5
termcolor==1.1.0
terminaltables==3.1.0
texttable==1.6.2
toml==0.10.0
typed-ast==1.4.0
unidiff==0.5.5
urllib3==1.22
utils==0.9.0
vine==1.3.0
wcwidth==0.1.7
websockets==4.0.1
Werkzeug==0.11.9
wrapt==1.11.2
WTForms==2.2.1
xmltodict==0.11.0
xxhash==1.3.0
zipp==0.5.2