title: Celery shutdown worker and retry the task
body: <p>I need to implement the following logic in celery task: if some condition is met, shutdown the current worker and retry the task.</p>
  
  <p>Tested on the sample task:</p>
  
  <pre class="lang-py prettyprint-override"><code>@app.task(bind=True, max_retries=1)
  def shutdown_and_retry(self, config):
      try:
          raise Exception('test exection')
      except Exception as exc:
          print('Retry {}/{}, task id {}'.format(self.request.retries, self.max_retries, self.request.id))
          app.control.shutdown(destination=[self.request.hostname])  # send shutdown signal to the current worker
          raise self.retry(exc=exc, countdown=5)
      print('Execute task id={} retries={}'.format(self.request.id, self.request.retries))
      return 'some result'
  </code></pre>
  
  <p>But it give strange results, steps:</p>
  
  <ol>
  <li>Run worker: <code>celery worker -Q test_queue -A test_worker -E -c 1 -n test_worker_1</code>.</li>
  <li>Push task to the "test_queue" queue.</li>
  <li>Worker caught it and shutdown. I opened the list of tasks in 'test_queue' in RabbitMQ and saw:
  
  <ul>
  <li>Original task submitted by publisher, retries = 0 (comes from app.control.shutdown() call);</li>
  <li>Copy of original task (with same id), retries = 1 (comes from self.retry() call).</li>
  </ul></li>
  <li>Then I started another worker to the same queue, it caught the task and shutdown also. But on Broker one more copy of the original task was pushed to the queue with the same id and retries = 1. So, I had 3 tasks in the queue. All next runs of workers gave + 1 new task to queue. Condition max_retries = 1 hasn't worked in this case.</li>
  </ol>
  
  <p>What I have tried:</p>
  
  <ol>
  <li>Set <code>task_reject_on_worker_lost = True</code> in celeryconfig.py and run the same task. Result: nothing changed.</li>
  <li>Leave only shutdown call in worker's task. Result: only original task is pushed back on each try (there is no tasks duplication), but it didn't count retries (always set to 0);</li>
  <li>Add <code>app.control.revoke(self.request.id)</code> before shutdown and retry calls in the worker (based on <a href="https://stackoverflow.com/questions/49348851/celery-shutdown-worker-after-particular-task">this</a>). Result: after the first try got the same (2 tasks in queue), but when I run second worker queue flushed and it didn't run anything. So, the task is lost and not retried.</li>
  </ol>
  
  <p>Is there a way to do not push back the original task to queue during <code>app.control.shutdown()</code> call? It seems that this is the root cause. Or could you please suggest another workaround which will allow to implement the right logic pointed above.</p>
  
  <p>Setup: RabbitMQ 3.8.2, celery 4.1.0, python 3.5.4</p>
  
  <p>Settings in celeryconfig.py:</p>
  
  <pre class="lang-py prettyprint-override"><code>task_acks_late = True
  task_acks_on_failure_or_timeout = True
  task_reject_on_worker_lost = False
  task_track_started = True
  worker_prefetch_multiplier = 1
  worker_disable_rate_limits = True
  </code></pre>