title: Apache Airflow tasks are stuck in a 'up_for_retry' state
body: <p>I've been setting up an airflow cluster on our system and previously it has been working. I'm not sure what I may have done to change this.</p>
  
  <p>I have a DAG which I want to run on a schedule. To make sure it's working I'd also like to trigger it manually. Neither of these seem to be working at the moment and no logs seem to be being written for the task instances. The only logs available are the airflow scheduler logs which generally look healthy.</p>
  
  <p>I am just constantly met with this message:</p>
  
  <p><code>Task is not ready for retry yet but will be retried automatically. Current date is 2018-12-12T11:34:46.978355+00:00 and task will be retried at 2018-12-12T11:35:08.093313+00:00.</code></p>
  
  <p>However, if I wait a little the exact same message is presented again except the times have moved forward a little. Therefore, it seems the task is never actually being retried.</p>
  
  <p>I am using a LocalExecutor and the task is an SSHOperator. Simplified code below. All it does is ssh onto another machine and start a bunch of application with a pre-determined directory structure.:</p>
  
  <pre><code>DB_INFO_FILE = 'info.json'
  START_SCRIPT = '/bin/start.sh'
  TIME_IN_PAST = timezone.convert_to_utc(datetime.today() - 
  timedelta(days=1))
  
  DEFAULT_ARGS = {
      'owner': 'airflow',
      'depends_on_past': False,
      'start_date': TIME_IN_PAST,
      'email': ['some_email@blah.com'],
      'email_on_failure': False,
      'email_on_retry': False,
      'retries': 1,
      'retry_delay': timedelta(minutes=1),
  }
  
  def _extract_instance_id(instance_string):
      return re.findall(r'\d+', instance_string)[0]
  
  def _read_file_as_json(file_name):
      with open(file_name) as open_file:
           return json.load(open_file)
  
  DB_INFO = _read_file_as_json(DB_INFO_FILE)
  CONFIG_CLIENT = ConfigDbClient(**DB_INFO)
  
  APP_DIRS = CONFIG_CLIENT.get_values('%my-app-info%')
  
  INSTANCE_START_SCRIPT_PATHS = {
      _extract_instance_id(instance_string): directory+START_SCRIPT
      for instance_string, directory in APP_DIRS.items()
      }
  
  # Create an ssh hook which refers to pre-existing connection information
  # setup and stored by airflow
  SSH_HOOK = SSHHook(ssh_conn_id='my-conn-id')
  
  # Create a DAG object to add tasks to
  DAG = DAG('my-dag-id',
            default_args=DEFAULT_ARGS)
  
  # Create a task for each app instance.
  for instance_id, start_script in INSTANCE_START_SCRIPT_PATHS.items():
      task = SSHOperator(
          task_id='run-script-{0}'.format(instance_id),
          command='bash {0}'.format(start_script),
          ssh_hook=SSH_HOOK,
          dag=DAG)
  </code></pre>
  
  <p>It works when I run the tasks individually, via the command line but not via the UI. It seems I can run tasks but I simply cannot trigger a DAG to run. I've tried many combinations of start_date s and interval schedules just to sanity check also.</p>
  
  <p>Any ideas?</p>
  
  <p>And yes, I am aware this question has been asked before and I have looked at all of them but not of the solutions have helped me.</p>