Fix handling of Lambda results
diff --git a/localstack/plugins.py b/localstack/plugins.py
index 955f28f1645d2..5562d72a153a2 100644
--- a/localstack/plugins.py
+++ b/localstack/plugins.py
@@ -9,7 +9,7 @@
 from localstack.services.kinesis import kinesis_listener, kinesis_starter
 from localstack.services.dynamodb import dynamodb_listener, dynamodb_starter
 from localstack.services.apigateway import apigateway_listener
-from localstack.services.stepfunctions import stepfunctions_starter
+from localstack.services.stepfunctions import stepfunctions_starter, stepfunctions_listener
 from localstack.services.cloudformation import cloudformation_listener, cloudformation_starter
 
 
@@ -69,7 +69,8 @@ def register_localstack_plugins():
         register_plugin(Plugin('cloudwatch',
             start=start_cloudwatch))
         register_plugin(Plugin('stepfunctions',
-            start=stepfunctions_starter.start_stepfunctions))
+            start=stepfunctions_starter.start_stepfunctions,
+            listener=stepfunctions_listener.UPDATE_STEPFUNCTIONS))
     except Exception as e:
         print('Unable to register plugins: %s' % e)
         raise e
diff --git a/localstack/services/awslambda/lambda_api.py b/localstack/services/awslambda/lambda_api.py
index bdd1a18dc1b80..48360a0ce37e3 100644
--- a/localstack/services/awslambda/lambda_api.py
+++ b/localstack/services/awslambda/lambda_api.py
@@ -13,7 +13,7 @@
 from io import BytesIO
 from datetime import datetime
 from six.moves import cStringIO as StringIO
-from flask import Flask, Response, jsonify, request, make_response
+from flask import Flask, Response, jsonify, request
 from localstack import config
 from localstack.services import generic_proxy
 from localstack.services.awslambda import lambda_executors
@@ -570,7 +570,7 @@ def format_func_details(func_details, version=None, always_add_version=False):
 @app.before_request
 def before_request():
     # fix to enable chunked encoding, as this is used by some Lambda clients
-    transfer_encoding = request.headers.get('Transfer-Encoding', None)
+    transfer_encoding = request.headers.get('Transfer-Encoding', '').lower()
     if transfer_encoding == 'chunked':
         request.environ['wsgi.input_terminated'] = True
 
@@ -815,23 +815,48 @@ def invoke_function(function):
     # Default invocation type is RequestResponse
     invocation_type = request.environ.get('HTTP_X_AMZ_INVOCATION_TYPE', 'RequestResponse')
 
+    def _create_response(result, status_code=200):
+        """ Create the final response for the given invocation result """
+        if isinstance(result, Response):
+            return result
+        details = {
+            'StatusCode': status_code,
+            'Payload': result,
+            'Headers': {}
+        }
+        if isinstance(result, dict):
+            for key in ('StatusCode', 'Payload', 'FunctionError'):
+                if result.get(key):
+                    details[key] = result[key]
+        # Try to parse parse payload as JSON
+        if isinstance(details['Payload'], (str, bytes)):
+            try:
+                details['Payload'] = json.loads(details['Payload'])
+            except Exception:
+                pass
+        # Set error headers
+        if details.get('FunctionError'):
+            details['Headers']['X-Amz-Function-Error'] = str(details['FunctionError'])
+        # Construct response object
+        response_obj = details['Payload']
+        if isinstance(response_obj, (dict, list)):
+            # Assume this is a JSON response
+            response_obj = jsonify(response_obj)
+        else:
+            details['Headers']['Content-Type'] = 'text/plain'
+        return response_obj, details['StatusCode'], details['Headers']
+
     if invocation_type == 'RequestResponse':
         result = run_lambda(asynchronous=False, func_arn=arn, event=data, context={}, version=qualifier)
-        if isinstance(result, dict):
-            return jsonify(result)
-        if result:
-            return result
-        return make_response('', 200)
+        return _create_response(result)
     elif invocation_type == 'Event':
         run_lambda(asynchronous=True, func_arn=arn, event=data, context={}, version=qualifier)
-        return make_response('', 202)
+        return _create_response('', status_code=202)
     elif invocation_type == 'DryRun':
         # Assume the dry run always passes.
-        return make_response('', 204)
-    else:
-        return error_response('Invocation type not one of: RequestResponse, Event or DryRun',
-                              code=400,
-                              error_type='InvalidParameterValueException')
+        return _create_response('', status_code=204)
+    return error_response('Invocation type not one of: RequestResponse, Event or DryRun',
+                          code=400, error_type='InvalidParameterValueException')
 
 
 @app.route('%s/event-source-mappings/' % PATH_ROOT, methods=['GET'])
diff --git a/localstack/services/awslambda/lambda_executors.py b/localstack/services/awslambda/lambda_executors.py
index 07e8e4b5abcab..04d0138a1e00e 100644
--- a/localstack/services/awslambda/lambda_executors.py
+++ b/localstack/services/awslambda/lambda_executors.py
@@ -5,7 +5,6 @@
 import logging
 import threading
 import subprocess
-# from datetime import datetime
 from multiprocessing import Process, Queue
 try:
     from shlex import quote as cmd_quote
@@ -75,9 +74,15 @@ def run_lambda_executor(self, cmd, env_vars={}, asynchronous=False):
             log_output = 'Lambda executed asynchronously'
         else:
             result, log_output = process.communicate()
-            result = to_str(result)
-            log_output = to_str(log_output)
+            result = to_str(result).strip()
+            log_output = to_str(log_output).strip()
             return_code = process.returncode
+            # Note: The user's code may have been logging to stderr, in which case the logs
+            # will be part of the "result" variable here. Hence, make sure that we extract
+            # only the *last* line of "result" and consider anything above that as log output.
+            if '\n' in result:
+                additional_logs, _, result = result.rpartition('\n')
+                log_output += '\n%s' % additional_logs
 
             if return_code != 0:
                 raise Exception('Lambda process returned error status code: %s. Output:\n%s' %
@@ -149,7 +154,8 @@ def _execute(self, func_arn, func_details, event, context=None, version=None, as
         # lambci writes the Lambda result to stdout and logs to stderr, fetch it from there!
         LOG.debug('Running lambda cmd: %s' % cmd)
         result, log_output = self.run_lambda_executor(cmd, environment, asynchronous)
-        LOG.debug('Lambda result / log output:\n%s\n>%s' % (result.strip(), log_output.strip().replace('\n', '\n> ')))
+        log_formatted = log_output.strip().replace('\n', '\n> ')
+        LOG.debug('Lambda %s result / log output:\n%s\n>%s' % (func_arn, result.strip(), log_formatted))
         return result, log_output
 
 
diff --git a/localstack/services/stepfunctions/stepfunctions_listener.py b/localstack/services/stepfunctions/stepfunctions_listener.py
index 259d487f8a77e..0ffb32e8df65a 100644
--- a/localstack/services/stepfunctions/stepfunctions_listener.py
+++ b/localstack/services/stepfunctions/stepfunctions_listener.py
@@ -6,10 +6,15 @@
 
 class ProxyListenerStepFunctions(ProxyListener):
 
-    def forward_request(self, method, path, data, headers):
-        LOG.debug('StepFunctions request:', method, path, data)
+    # TODO: listener methods currently disabled
+
+    def forward_request_DISABLED(self, method, path, data, headers):
+        LOG.debug('StepFunctions request: %s %s %s', method, path, data)
         return True
 
+    def return_response_DISABLED(self, method, path, data, headers, response):
+        LOG.debug('StepFunctions response: %s %s %s %s', method, path, response.status_code, response.content)
+
 
 # instantiate listener
 UPDATE_STEPFUNCTIONS = ProxyListenerStepFunctions()
diff --git a/requirements.txt b/requirements.txt
index 533120547a99a..d2a8d5379b907 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -12,7 +12,7 @@ docopt>=0.6.2
 elasticsearch==6.2.0
 flake8>=3.6.0
 flake8-quotes>=0.11.0
-flask==0.12.4
+flask==1.0.2
 flask-cors==3.0.3
 flask_swagger==0.2.12
 jsonpath-rw==1.4.0
diff --git a/tests/integration/test_stepfunction.py b/tests/integration/test_stepfunction.py
index 85e83b383515f..0a3432470c91f 100644
--- a/tests/integration/test_stepfunction.py
+++ b/tests/integration/test_stepfunction.py
@@ -9,13 +9,19 @@
 
 THIS_FOLDER = os.path.dirname(os.path.realpath(__file__))
 TEST_LAMBDA_PYTHON = os.path.join(THIS_FOLDER, 'lambdas', 'lambda_environment.py')
-TEST_LAMBDA_NAME = 'lambda_sfn_1'
+TEST_LAMBDA_NAME_1 = 'lambda_sfn_1'
+TEST_LAMBDA_NAME_2 = 'lambda_sfn_2'
 STATE_MACHINE_NAME = 'test_sm_1'
 STATE_MACHINE_DEF = {
     'Comment': 'Hello World example',
     'StartAt': 'step1',
     'States': {
         'step1': {
+            'Type': 'Task',
+            'Resource': '__tbd__',
+            'Next': 'step2'
+        },
+        'step2': {
             'Type': 'Task',
             'Resource': '__tbd__',
             'End': True
@@ -38,14 +44,20 @@ def setUpClass(cls):
             runtime=LAMBDA_RUNTIME_PYTHON36
         )
         testutil.create_lambda_function(
-            func_name=TEST_LAMBDA_NAME,
+            func_name=TEST_LAMBDA_NAME_1,
+            zip_file=zip_file,
+            runtime=LAMBDA_RUNTIME_PYTHON36
+        )
+        testutil.create_lambda_function(
+            func_name=TEST_LAMBDA_NAME_2,
             zip_file=zip_file,
             runtime=LAMBDA_RUNTIME_PYTHON36
         )
 
     @classmethod
     def tearDownClass(cls):
-        cls.lambda_client.delete_function(FunctionName=TEST_LAMBDA_NAME)
+        cls.lambda_client.delete_function(FunctionName=TEST_LAMBDA_NAME_1)
+        cls.lambda_client.delete_function(FunctionName=TEST_LAMBDA_NAME_2)
 
     def test_create_run_state_machine(self):
         state_machines_before = self.sfn_client.list_state_machines()['stateMachines']
@@ -53,8 +65,10 @@ def test_create_run_state_machine(self):
         # create state machine
         role_arn = aws_stack.role_arn('sfn_role')
         definition = clone(STATE_MACHINE_DEF)
-        lambda_arn = aws_stack.lambda_function_arn(TEST_LAMBDA_NAME)
-        definition['States']['step1']['Resource'] = lambda_arn
+        lambda_arn_1 = aws_stack.lambda_function_arn(TEST_LAMBDA_NAME_1)
+        lambda_arn_2 = aws_stack.lambda_function_arn(TEST_LAMBDA_NAME_2)
+        definition['States']['step1']['Resource'] = lambda_arn_1
+        definition['States']['step2']['Resource'] = lambda_arn_2
         definition = json.dumps(definition)
         result = self.sfn_client.create_state_machine(
             name=STATE_MACHINE_NAME, definition=definition, roleArn=role_arn)
@@ -70,10 +84,11 @@ def test_create_run_state_machine(self):
         self.assertTrue(result.get('executionArn'))
 
         def check_invocations():
-            assert lambda_api.LAMBDA_EXECUTOR.function_invoke_times[lambda_arn]
+            assert lambda_api.LAMBDA_EXECUTOR.function_invoke_times[lambda_arn_1]
+            assert lambda_api.LAMBDA_EXECUTOR.function_invoke_times[lambda_arn_2]
 
         # assert that the lambda has been invoked by the SM execution
-        retry(check_invocations, sleep=0.6, retries=15)
+        retry(check_invocations, sleep=0.7, retries=20)
 
         # clean up
         self.sfn_client.delete_state_machine(stateMachineArn=sm_arn)

Fix handling of Lambda invocation results.

* Update `flask` to latest version (fixes an issue with chunked transfer encoding in Lambda API)
* Fix Step Functions integration test (previously were hanging as soon as more than one state was defined in the state machine)
* Return `X-Amz-Function-Error` header in Lambda API