**Apache Airflow version**:
2.0 (master)

**Environment**:

```
./breeze --python 3.7 --db-reset restart --integration redis
```

**What happened**:

Running `airflow celery worker -D` spawns celery worker (`ps -aux`) but the worker does not consume any tasks 👀 
Worker logs show:
```
2020-10-17 22:57:20,845 INFO - Connected to redis://redis:6379/0
2020-10-17 22:57:20,858 INFO - mingle: searching for neighbors
2020-10-17 22:57:20,865 WARNING - consumer: Connection to broker lost. Trying to re-establish the connection...
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/celery/worker/consumer/consumer.py", line 318, in start
    blueprint.start(self)
  File "/usr/local/lib/python3.7/site-packages/celery/bootsteps.py", line 119, in start
    step.start(parent)
  File "/usr/local/lib/python3.7/site-packages/celery/worker/consumer/mingle.py", line 40, in start
    self.sync(c)
  File "/usr/local/lib/python3.7/site-packages/celery/worker/consumer/mingle.py", line 44, in sync
    replies = self.send_hello(c)
  File "/usr/local/lib/python3.7/site-packages/celery/worker/consumer/mingle.py", line 57, in send_hello
    replies = inspect.hello(c.hostname, our_revoked._data) or {}
  File "/usr/local/lib/python3.7/site-packages/celery/app/control.py", line 156, in hello
    return self._request('hello', from_node=from_node, revoked=revoked)
  File "/usr/local/lib/python3.7/site-packages/celery/app/control.py", line 106, in _request
    pattern=self.pattern, matcher=self.matcher,
  File "/usr/local/lib/python3.7/site-packages/celery/app/control.py", line 480, in broadcast
    limit, callback, channel=channel,
  File "/usr/local/lib/python3.7/site-packages/kombu/pidbox.py", line 352, in _broadcast
    channel=chan)
  File "/usr/local/lib/python3.7/site-packages/kombu/pidbox.py", line 391, in _collect
    self.connection.drain_events(timeout=timeout)
  File "/usr/local/lib/python3.7/site-packages/kombu/connection.py", line 324, in drain_events
    return self.transport.drain_events(self.connection, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/kombu/transport/virtual/base.py", line 963, in drain_events
    get(self._deliver, timeout=timeout)
  File "/usr/local/lib/python3.7/site-packages/kombu/transport/redis.py", line 369, in get
    self._register_BRPOP(channel)
  File "/usr/local/lib/python3.7/site-packages/kombu/transport/redis.py", line 308, in _register_BRPOP
    self._register(*ident)
  File "/usr/local/lib/python3.7/site-packages/kombu/transport/redis.py", line 292, in _register
    self.poller.register(sock, self.eventflags)
  File "/usr/local/lib/python3.7/site-packages/kombu/utils/eventio.py", line 67, in register
    self._epoll.register(fd, events)
OSError: [Errno 22] Invalid argument
2020-10-17 22:57:20,867 CRITICAL - Frequent restarts detected: RestartFreqExceeded('5 in 1s')
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/celery/worker/consumer/consumer.py", line 312, in start
    self._restart_state.step()
  File "/usr/local/lib/python3.7/site-packages/billiard/common.py", line 165, in step
    raise self.RestartFreqExceeded("%r in %rs" % (R, self.maxT))
billiard.exceptions.RestartFreqExceeded: 5 in 1s
```

**What you expected to happen**:

I expect that demonized worker will consume Airflow tasks.

**How to reproduce it**:
```
./breeze --python 3.7 --db-reset restart --integration redis
airflow scheduler -D
airflow webserver -D -w 1
airflow celery worker -c 1 -D
```
Then trigger any DAG from webui.


**Anything else we need to know**:

Worked before: #11336

airflow/executors/celery_executor.py
Tuple[str, Union[str, ExceptionWithTraceback], Any]:
    """
    Fetch and return the state of the given celery task. The scope of this function is
    global so that it can be called by subprocesses in the pool.
    :param async_result: a tuple of the Celery task key and the async Celery object used
        to fetch the task's state
    :type async_result: tuple(str, celery.result.AsyncResult)
    :return: a tuple of the Celery task key and the Celery state and the celery info
        of the task
    :rtype: tuple[str, str, str]
    """
    try:
        with timeout(seconds=OPERATION_TIMEOUT):
            
            
            info = async_result.info if hasattr(async_result, 'info') else None
            return async_result.task_id, async_result.state, info
    except Exception as e:  
        exception_traceback = f"Celery Task ID: {async_result}\n{traceback.format_exc()}"
        return async_result.task_id, ExceptionWithTraceback(e, exception_traceback), None
def _tasks_list_to_task_ids(async_tasks) -> Set[str]:
    return {a.task_id for a in async_tasks}
class BulkStateFetcher(LoggingMixin):