EMR DAG terminates before all steps are completed

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.

Through some digging, I can see that Airflow "pokes" steps to see if they are still running. In this instance, it seems to think that it's "Successful" only when 1 step is finished.

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.

The DAG is below

    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',
            "maximizeResourceAllocation":"true"},
        '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="{{ task_instance.xcom_pull('create_job_flow', key='return_value') }}",
        aws_conn_id='aws_default',
        steps=SPARK_TEST_STEPS,
        dag=dag
    )

    step_checker = EmrStepSensor(
        task_id='watch_step',
        job_flow_id="{{ task_instance.xcom_pull('create_job_flow', 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_flow', key='return_value') }}",
        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)