**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
import signal
            signal.signal(signal.SIGINT, signal.SIG_DFL)
            signal.signal(signal.SIGTERM, signal.SIG_DFL)
        with Pool(processes=num_processes, initializer=reset_signals) as send_pool:
            key_and_async_results = send_pool.map(
                send_task_to_executor,
                task_tuples_to_send,
                chunksize=chunksize)
        return key_and_async_results
    def sync(self) -> None:
        if not self.tasks:
            self.log.debug("No task to query celery, skipping sync")
            return
        self.update_all_task_states()
        if self.adopted_task_timeouts:
            self._check_for_stalled_adopted_tasks()
    def _check_for_stalled_adopted_tasks(self):
        """
        See if any of the tasks we adopted from another Executor run have not
        progressed after the configured timeout.
        If they haven't, they likely never made it to Celery, and we should