Kombu: "Error 32 while writing to socket. Broken pipe" After upgrading to kombu 4.4.0 and redis 3.2.0

[alexandre-paroissien]

I was previously on kombu 4.3.0 and redis 2.10.6, and after upgrading to kombu 4.4.0 and redis 3.2.0 (manually cf issue #1016 ) I noticed an issue on one of my Django REST endpoint:

OperationalError: Error 32 while writing to socket. Broken pipe.

I am not sure if it's an actual issue in the new kombu / redis, or if I had an issue in my code that was silenced in the previous kombu / redis version

The endpoint is heavily used, and never had any issue, this issue popped up a few times since I upgraded, but the endpoint still works most of the times

Any ideas?

Edit: Celery version is 4.2.1

Here is the stack trace:

File "/app/.heroku/python/lib/python3.6/site-packages/redis/connection.py" in send_packed_command
  600.                 self._sock.sendall(item)

During handling of the above exception ([Errno 32] Broken pipe), another exception occurred:

File "/app/.heroku/python/lib/python3.6/site-packages/kombu/connection.py" in _reraise_as_library_errors
  431.             yield

File "/app/.heroku/python/lib/python3.6/site-packages/celery/app/base.py" in send_task
  744.                     self.backend.on_task_call(P, task_id)

File "/app/.heroku/python/lib/python3.6/site-packages/celery/backends/redis.py" in on_task_call
  265.             self.result_consumer.consume_from(task_id)

File "/app/.heroku/python/lib/python3.6/site-packages/celery/backends/redis.py" in consume_from
  126.         self._consume_from(task_id)

File "/app/.heroku/python/lib/python3.6/site-packages/celery/backends/redis.py" in _consume_from
  132.             self._pubsub.subscribe(key)

File "/app/.heroku/python/lib/python3.6/site-packages/redis/client.py" in subscribe
  3096.         ret_val = self.execute_command('SUBSCRIBE', *iterkeys(new_channels))

File "/app/.heroku/python/lib/python3.6/site-packages/redis/client.py" in execute_command
  3009.         self._execute(connection, connection.send_command, *args)

File "/app/.heroku/python/lib/python3.6/site-packages/redis/client.py" in _execute
  3013.             return command(*args)

File "/app/.heroku/python/lib/python3.6/site-packages/newrelic/hooks/datastore_redis.py" in _nr_Connection_send_command_wrapper_
  129.         return wrapped(*args, **kwargs)

File "/app/.heroku/python/lib/python3.6/site-packages/redis/connection.py" in send_command
  620.         self.send_packed_command(self.pack_command(*args))

File "/app/.heroku/python/lib/python3.6/site-packages/redis/connection.py" in send_packed_command
  613.                                   (errno, errmsg))

During handling of the above exception (Error 32 while writing to socket. Broken pipe.), another exception occurred:

File "/app/.heroku/python/lib/python3.6/site-packages/django/core/handlers/exception.py" in inner
  34.             response = get_response(request)

File "/app/.heroku/python/lib/python3.6/site-packages/django/core/handlers/base.py" in _get_response
  126.                 response = self.process_exception_by_middleware(e, request)

File "/app/.heroku/python/lib/python3.6/site-packages/django/core/handlers/base.py" in _get_response
  124.                 response = wrapped_callback(request, *callback_args, **callback_kwargs)

File "/app/.heroku/python/lib/python3.6/site-packages/newrelic/hooks/framework_django.py" in wrapper
  544.                 return wrapped(*args, **kwargs)

File "/app/.heroku/python/lib/python3.6/site-packages/django/views/decorators/csrf.py" in wrapped_view
  54.         return view_func(*args, **kwargs)

File "/app/.heroku/python/lib/python3.6/site-packages/django/views/generic/base.py" in view
  68.             return self.dispatch(request, *args, **kwargs)

File "/app/.heroku/python/lib/python3.6/site-packages/newrelic/hooks/component_djangorestframework.py" in _nr_wrapper_APIView_dispatch_
  46.         return wrapped(*args, **kwargs)

File "/app/.heroku/python/lib/python3.6/site-packages/rest_framework/views.py" in dispatch
  495.             response = self.handle_exception(exc)

File "/app/.heroku/python/lib/python3.6/site-packages/newrelic/hooks/component_djangorestframework.py" in _handle_exception_wrapper
  53.         return wrapped(*args, **kwargs)

File "/app/.heroku/python/lib/python3.6/site-packages/rest_framework/views.py" in handle_exception
  455.             self.raise_uncaught_exception(exc)

File "/app/.heroku/python/lib/python3.6/site-packages/rest_framework/views.py" in dispatch
  492.             response = handler(request, *args, **kwargs)

File "/app/.heroku/python/lib/python3.6/site-packages/rest_framework/decorators.py" in handler
  55.             return func(*args, **kwargs)

File "/app/candidate/views.py" in AddLinkedinView
  698.     create_candidate_skills.delay(candidate.pk, candidate_skills, source)

File "/app/.heroku/python/lib/python3.6/site-packages/celery/app/task.py" in delay
  408.         return self.apply_async(args, kwargs)

File "/app/.heroku/python/lib/python3.6/site-packages/celery/app/task.py" in apply_async
  535.             **options

File "/app/.heroku/python/lib/python3.6/site-packages/celery/app/base.py" in send_task
  745.                 amqp.send_task_message(P, name, message, **options)

File "/app/.heroku/python/lib/python3.6/contextlib.py" in __exit__
  99.                 self.gen.throw(type, value, traceback)

File "/app/.heroku/python/lib/python3.6/site-packages/kombu/connection.py" in _reraise_as_library_errors
  436.                     sys.exc_info()[2])

File "/app/.heroku/python/lib/python3.6/site-packages/vine/five.py" in reraise
  194.             raise value.with_traceback(tb)

File "/app/.heroku/python/lib/python3.6/site-packages/kombu/connection.py" in _reraise_as_library_errors
  431.             yield

File "/app/.heroku/python/lib/python3.6/site-packages/celery/app/base.py" in send_task
  744.                     self.backend.on_task_call(P, task_id)

File "/app/.heroku/python/lib/python3.6/site-packages/celery/backends/redis.py" in on_task_call
  265.             self.result_consumer.consume_from(task_id)

File "/app/.heroku/python/lib/python3.6/site-packages/celery/backends/redis.py" in consume_from
  126.         self._consume_from(task_id)

File "/app/.heroku/python/lib/python3.6/site-packages/celery/backends/redis.py" in _consume_from
  132.             self._pubsub.subscribe(key)

File "/app/.heroku/python/lib/python3.6/site-packages/redis/client.py" in subscribe
  3096.         ret_val = self.execute_command('SUBSCRIBE', *iterkeys(new_channels))

File "/app/.heroku/python/lib/python3.6/site-packages/redis/client.py" in execute_command
  3009.         self._execute(connection, connection.send_command, *args)

File "/app/.heroku/python/lib/python3.6/site-packages/redis/client.py" in _execute
  3013.             return command(*args)

File "/app/.heroku/python/lib/python3.6/site-packages/newrelic/hooks/datastore_redis.py" in _nr_Connection_send_command_wrapper_
  129.         return wrapped(*args, **kwargs)

File "/app/.heroku/python/lib/python3.6/site-packages/redis/connection.py" in send_command
  620.         self.send_packed_command(self.pack_command(*args))

File "/app/.heroku/python/lib/python3.6/site-packages/redis/connection.py" in send_packed_command
  613.                                   (errno, errmsg))

[iven]

I confirm this issue still exists in Kombu 4.5.0 and redis 3.2.1. Random broken pipe errors, as a result, my tasks random fail to start.

[vpol]

Any updates?

[alexandre-paroissien]

I am still having the issues in the latest branches of celery / kombu / redis

So I am staying on the previous branches for now (celery 4.2 / kombu 4.3 / redispy 2.10)

It seems the issue is coming from redispy 3.x branch, and using this redispy branch is required in new celery / kombu branches (celery 4.3+ / kombu 4.4)

You can check the redis issue status there
https://github.com/andymccurdy/redis-py/issues/1140

[vpol]

I've seen It before, but it looks like they are not going to fix it (as it's not an issue for them).

[ghost]

This is a big issue for me as well.

It looks like this was fixed on the redis-py side, but pubsub.check_health() needs to be added on the celery side.

https://github.com/andymccurdy/redis-py/issues/1140#issuecomment-515634842

[auvipy]

can you come with a PR @frogger72

[zishenWang]

There is the same problem seem like the result from redis backend can not to get,althourgh it say the exception has been ignore,and the worker still working for my flowing task.But this problem should course my webserver out of memory if i don't restart in schedule.
ps:
redis Version: 3.3.11
celery Version: 4.4.0
kombu: 4.6.7