AIRFLOW-3297 added possible states to the EMR step sensor
diff --git a/airflow/contrib/sensors/emr_step_sensor.py b/airflow/contrib/sensors/emr_step_sensor.py
index 65bb7c95b1303..315fdf7427d31 100644
--- a/airflow/contrib/sensors/emr_step_sensor.py
+++ b/airflow/contrib/sensors/emr_step_sensor.py
@@ -32,8 +32,8 @@ class EmrStepSensor(EmrBaseSensor):
     :type step_id: str
     """
 
-    NON_TERMINAL_STATES = ['PENDING', 'RUNNING', 'CONTINUE']
-    FAILED_STATE = ['CANCELLED', 'FAILED']
+    NON_TERMINAL_STATES = ['PENDING', 'RUNNING', 'CONTINUE', 'CANCEL_PENDING']
+    FAILED_STATE = ['CANCELLED', 'FAILED', 'INTERRUPTED']
     template_fields = ['job_flow_id', 'step_id']
     template_ext = ()
 
diff --git a/tests/contrib/sensors/test_emr_step_sensor.py b/tests/contrib/sensors/test_emr_step_sensor.py
index 2a105a9393d21..08cb8d1aeac52 100644
--- a/tests/contrib/sensors/test_emr_step_sensor.py
+++ b/tests/contrib/sensors/test_emr_step_sensor.py
@@ -83,6 +83,35 @@
     }
 }
 
+DESCRIBE_JOB_STEP_INTERRUPTED_RETURN = {
+    'ResponseMetadata': {
+        'HTTPStatusCode': 200,
+        'RequestId': '8dee8db2-3719-11e6-9e20-35b2f861a2a6'
+    },
+    'Step': {
+        'ActionOnFailure': 'CONTINUE',
+        'Config': {
+            'Args': [
+                '/usr/lib/spark/bin/run-example',
+                'SparkPi',
+                '10'
+            ],
+            'Jar': 'command-runner.jar',
+            'Properties': {}
+        },
+        'Id': 's-VK57YR1Z9Z5N',
+        'Name': 'calculate_pi',
+        'Status': {
+            'State': 'INTERRUPTED',
+            'StateChangeReason': {},
+            'Timeline': {
+                'CreationDateTime': datetime(2016, 6, 20, 19, 0, 18, tzinfo=tzlocal()),
+                'StartDateTime': datetime(2016, 6, 20, 19, 2, 34, tzinfo=tzlocal())
+            }
+        }
+    }
+}
+
 DESCRIBE_JOB_STEP_COMPLETED_RETURN = {
     'ResponseMetadata': {
         'HTTPStatusCode': 200,
@@ -158,6 +187,17 @@ def test_step_cancelled(self):
         with patch('boto3.session.Session', self.boto3_session_mock):
             self.assertRaises(AirflowException, self.sensor.execute, None)
 
+    def test_step_interrupted(self):
+        self.emr_client_mock.describe_step.side_effect = [
+            DESCRIBE_JOB_STEP_RUNNING_RETURN,
+            DESCRIBE_JOB_STEP_INTERRUPTED_RETURN
+        ]
+
+        self.boto3_client_mock = MagicMock(return_value=self.emr_client_mock)
+
+        with patch('boto3.session.Session', self.boto3_session_mock):
+            self.assertRaises(AirflowException, self.sensor.execute, None)
+
 
 if __name__ == '__main__':
     unittest.main()

Make sure you have checked _all_ steps below.

### Jira

- [ ] My PR addresses [AIRFLOW-3297](https://issues.apache.org/jira/browse/AIRFLOW-3297)

### Description

- [ ] This PR implements correct behaviour to EMR step states. Previously, the states 'PENDING_CANCELLED' and 'INTERRUPTED' resulted in the Airflow task being marked as successful. This PR fixed this by mentioning those states as NON_TERMINAL_STATES and as FAILED_STATE, accordingly.

### Tests

- [ ] My PR adds the following unit tests - test_step_interrupted

### Code Quality

- [ ] Passes `flake8`



