title: EMR DAG terminates before all steps are completed
body: <p>I am using the EMR CreateJobFlow, AddSteps,StepSensor and TerminateJobFlow operators in my DAG to start up an EMR cluster, add steps (2 spark apps, and dist-cp), and terminate when all steps are completed or 1 fails. I am able to do this when I have a 2 step DAG (1st is Spark app, 2nd is dist-cp), however, when I have 2 spark apps, the cluster successfully runs the 1st step, and terminates without moving on to the 2nd and 3rd steps.</p>
  <p>Through some digging, I can see that Airflow &quot;pokes&quot; steps to see if they are still running. In this instance, it seems to think that it's &quot;Successful&quot; only when 1 step is finished.</p>
  <p>My spark apps are fairly simple. The first creates and writes a dataframe to local HDFS. The 2nd reads the data from HDFS and joins to another data set and writes back to HDFS. The third step is s3-dist-cp to copy the data from HDFS to s3. All 3 steps can be successfully run within Spark-Shell interactively, or as Spark-Submit jobs. I have also cloned the EMR cluster myself (without airflow) and saw all steps succeed without any errors, so EMR and Spark are not the issue here.</p>
  <p>The DAG is below</p>
  <pre class="lang-py prettyprint-override"><code>
      from datetime import timedelta
  
      import airflow
      from airflow import DAG
      from airflow.contrib.operators.emr_create_job_flow_operator \
          import EmrCreateJobFlowOperator
      from airflow.contrib.operators.emr_add_steps_operator \
          import EmrAddStepsOperator
      from airflow.contrib.sensors.emr_step_sensor import EmrStepSensor
      from airflow.contrib.operators.emr_terminate_job_flow_operator \
          import EmrTerminateJobFlowOperator
  
      DEFAULT_ARGS = {
          'owner': 'airflow',
          'depends_on_past': False,
          'start_date': airflow.utils.dates.days_ago(2)   
      }
  
      SPARK_TEST_STEPS = [
          {
              'Name': 'monthly_agg',
              'ActionOnFailure': 'CONTINUE',
              'HadoopJarStep': {
                  'Jar': 'command-runner.jar',
                  'Args': ['spark-submit',
                            '--deploy-mode',
                             'cluster',
                            '--class' ,
                            'AggApp',
                            's3://jar1.jar' ]
                              }
          },   
          {
              'Name': 'monthly_agg2',
              'ActionOnFailure': 'CONTINUE',
              'HadoopJarStep': {
                  'Jar': 'command-runner.jar',
                  'Args': ['spark-submit',
                            '--deploy-mode',
                             'cluster',
                            '--class' ,
                            'SimpleApp',
                            's3:/jar2.jar' ]
                              }
          },  
  
          {
              'Name': 'copy-data',
              'ActionOnFailure': 'CONTINUE',
              'HadoopJarStep': {
                  'Jar': 'command-runner.jar',
                  'Args': ['s3-dist-cp',
                            '--src',
                            '/tempo',
                            '--dest',
                             's3://mydata/'
                          ]
                              }
          }
      ]
  
      JOB_FLOW_OVERRIDES = {
      'Instances': {'Ec2SubnetId': 'subnet-mysubnetid', 
          'InstanceGroups': [
              {
                  'Name': 'Master nodes',
                  'Market': 'ON_DEMAND',
                  'InstanceRole': 'MASTER',
                  'InstanceType': 'r4.2xlarge',
                  'InstanceCount': 1
              },
              {
                  'Name': 'Slave nodes',
                  'Market': 'ON_DEMAND',
                  'InstanceRole': 'CORE',
                  'InstanceType': 'r4.2xlarge',
                  'InstanceCount': 8,
                  'EbsConfiguration': {'EbsBlockDeviceConfigs':[{'VolumeSpecification':{'SizeInGB':128,'VolumeType':'gp2'},'VolumesPerInstance':1}],'EbsOptimized':True}
              }
          ]},
          'Name':'airflow-monthly_agg_custom',
          'Configurations': [
          {
          'Classification':'spark-defaults','Properties':
              {'spark.sql.crossJoin.enabled':'true',
              'spark.serializer':'org.apache.spark.serializer.KryoSerializer',
              'spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version':'2',
              &quot;maximizeResourceAllocation&quot;:&quot;true&quot;},
          'Configurations':[]
          },
          {
          'Classification':'spark-hive-site','Properties':
          {'hive.metastore.client.factory.class':'com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory'}, 
          'Configurations':[]
          }
          ]}
  
      dag = DAG(
          'monthly_agg_custom',
          default_args=DEFAULT_ARGS,
          dagrun_timeout=timedelta(hours=4),
          schedule_interval='@once'
      )
  
      cluster_creator = EmrCreateJobFlowOperator(
          task_id='create_job_flow',
          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=&quot;{{ task_instance.xcom_pull('create_job_flow', key='return_value') }}&quot;,
          aws_conn_id='aws_default',
          steps=SPARK_TEST_STEPS,
          dag=dag
      )
  
      step_checker = EmrStepSensor(
          task_id='watch_step',
          job_flow_id=&quot;{{ task_instance.xcom_pull('create_job_flow', key='return_value') }}&quot;,
          step_id=&quot;{{ task_instance.xcom_pull('add_steps', key='return_value')[0] }}&quot;,
          aws_conn_id='aws_default',
          dag=dag
      )
  
      cluster_remover = EmrTerminateJobFlowOperator(
          task_id='remove_cluster',
          job_flow_id=&quot;{{ task_instance.xcom_pull('create_job_flow', key='return_value') }}&quot;,
          aws_conn_id='aws_default',
          dag=dag
      )
  
      cluster_creator.set_downstream(step_adder)
      step_adder.set_downstream(step_checker)
      step_checker.set_downstream(cluster_remover)
  </code></pre>