Worker: Crash hard on internal task errors (Issue #3539)

celery/concurrency/asynpool.py

@@ -1150,11 +1150,12 @@ class AsynPool(_pool.Pool):
     def _setup_queues(self):
         # this is only used by the original pool that used a shared
         # queue for all processes.
+        self._quick_put = None
 
-        # these attributes makes no sense for us, but we'll still
-        # have to initialize them.
+        # these attributes are unused by this class, but we'll still
+        # have to initialize them for compatibility.
         self._inqueue = self._outqueue = \
-            self._quick_put = self._quick_get = self._poll_result = None
+            self._quick_get = self._poll_result = None
 
     def process_flush_queues(self, proc):
         """Flush all queues.

celery/worker/consumer/consumer.py

@@ -561,9 +561,6 @@ class Consumer(object):
                     return on_invalid_task(payload, message, exc)
                 except MemoryError:
                     raise
-                except Exception as exc:  # pylint: disable=broad-except
-                    # XXX handle as internal error?
-                    return on_invalid_task(payload, message, exc)
 
         return on_task_received
 

celery/worker/worker.py

@@ -230,9 +230,6 @@ class WorkController(object):
                 self._quick_release()   # Issue 877
             except AttributeError:
                 pass
-        except Exception as exc:
-            logger.critical('Internal error: %r\n%s',
-                            exc, traceback.format_exc(), exc_info=True)
 
     def signal_consumer_close(self):
         try: