Version info

• celery: 4.2.0
• kombu: 4.2.0
• redis-py: 2.10.6
• gevent: 1.2.2
• redis server: 3.2.11

Steps to reproduce

Celery configuration: broker is rabbitmq, results backend is local redis (password protected). Both broker & client are gevented.

1. start redis server
2. start a celery worker
3. call task.apply_async() once (tasks executes and output is returned)
4. restart redis server
5. call task.apply_async() again (caller hangs forever)

Expected behavior

• During server downtime: apply_async().get() is expected to raise an error (ConnectionError: Error while reading from socket: ('Connection closed by server.',))
• After server is up again: apply_async().get() is expected to return to normal

Actual behavior

• During server downtime: apply_async().get() sometimes raises an error and sometimes just hangs
• After server is up again: apply_async().get() always hangs. Celery worker receives task request (over rabbitmq broker), executes it. Output is correctly stored in redis under celery-task-meta-TASKID. Caller never actually collects output and returns.

Additional information

• Issue looks similar to Celery hangs indefinitely (instead of failing) if redis is not started #4556, however this one is not related to worker bootstrapping but rather to client hanging after disconnection.
• Issue also reproduces (in fact, first occurred for us) when celery's redis client disconnects from the server due to network / proxy issues (in our case - middle-man HA proxy drops the connections).