Pool.flush may not exist

celery/concurrency/prefork.py

@@ -128,7 +128,7 @@ class TaskPool(BasePool):
         self.terminate_job = P.terminate_job
         self.grow = P.grow
         self.shrink = P.shrink
-        self.flush = P.flush
+        self.flush = getattr(P, 'flush', None)  # FIXME add to billiard
         self.restart = P.restart
 
     def did_start_ok(self):

celery/worker/consumer.py

@@ -332,7 +332,7 @@ class Consumer(object):
         if self.timer:
             self.timer.clear()
         reserved_requests.clear()
-        if self.pool:
+        if self.pool and self.pool.flush:
             self.pool.flush()
 
     def connect(self):