title: Airflow emr cluster termination being called before step completion
body: <p>I am launching a cluster on EMR and submiting some steps on it using airflow.</p>
  
  <p><strong>What i want:</strong></p>
  
  <p>I want to terminate my cluster after all the steps added through <code>EmrAddStepsOperator</code> are complete</p>
  
  <p><strong>What i have tried:</strong></p>
  
  <p>I have tried using together <code>EmrStepSensor</code> and <code>EmrTerminateJobFlowOperator</code>, but my spark step gets cancelled and cluster terminates without completing all the steps</p>
  
  <p>Anyone please suggest how to do it properly. here is my code</p>
  
  <pre><code>dag = DAG('emr_job_flow_automatic_steps_17',
           default_args=default_args,
           schedule_interval="@daily",
           max_active_runs=1,
           catchup=True,
  )
  
  
  upload_to_S3_task = PythonOperator(
     task_id='upload_to_S3',
     python_callable=upload_file_to_S3,
     op_kwargs={
         'filename': '/home/ab/projects/test.py',
         'key': 'test.py',
         'bucket_name': 'dep-buck',
     },
     dag=dag)
  
  cluster_creator = EmrCreateJobFlowOperator(
     task_id='create_job_flow2',
     job_flow_overrides=JOB_FLOW_OVERRIDES,
     aws_conn_id='aws_default',
     emr_conn_id='emr_default',
     dag=dag
  )
  
  step_adder = EmrAddStepsOperator(
     task_id='add_steps',
     job_flow_id="{{ task_instance.xcom_pull('create_job_flow2', key='return_value') }}",
     aws_conn_id='aws_default',
     steps=step,
     dag=dag
  )
  step_checker = EmrStepSensor(
     task_id='watch_step',
     job_flow_id="{{ task_instance.xcom_pull('create_job_flow2', key='return_value') }}",
     step_id="{{ task_instance.xcom_pull('add_steps', key='return_value')[0] }}",
     aws_conn_id='aws_default',
     dag=dag
  )
  
  cluster_remover = EmrTerminateJobFlowOperator(
     task_id='remove_cluster',
     job_flow_id="{{ task_instance.xcom_pull('create_job_flow2', key='return_value') }}",
     aws_conn_id='aws_default',
     dag=dag
  )
  
  upload_to_S3_task &gt;&gt; cluster_creator &gt;&gt; step_adder &gt;&gt; step_checker &gt;&gt; cluster_remover
  </code></pre>
  
  <p><strong>Other Questions on stackoverflow:</strong></p>
  
  <p>There is one similar question on stackoverflow but it has not been answered properly (using EmrTerminateJobFlowOperator)</p>