localstack · alexrashed · Aug 30, 2022 · May 24, 2022 · Jul 27, 2022 · Aug 16, 2022
diff --git a/localstack/aws/api/kinesis/__init__.py b/localstack/aws/api/kinesis/__init__.py
diff --git a/localstack/aws/forwarder.py b/localstack/aws/forwarder.py
@@ -74,7 +74,12 @@ def _forward_request(context, service_request: ServiceRequest = None) -> Service
                 parameters=service_request,
                 region=context.region,
             )
-            local_context.request.headers.update(context.request.headers)
+            # update the newly created context with non-payload specific request headers (the payload can differ from
+            # the original request, f.e. it could be JSON encoded now while the initial request was CBOR encoded)
+            headers = Headers(context.request.headers)
+            headers.pop("Content-Type", None)
+            headers.pop("Content-Length", None)
+            local_context.request.headers.update(headers)
             context = local_context
         return forward_request(context, forward_url_getter)
 

diff --git a/localstack/aws/handlers/service.py b/localstack/aws/handlers/service.py
@@ -150,7 +150,7 @@ def create_not_implemented_response(self, context):
         message = f"no handler for operation '{operation_name}' on service '{service_name}'"
         error = CommonServiceException("InternalFailure", message, status_code=501)
         serializer = create_serializer(context.service)
-        return serializer.serialize_error_to_response(error, operation)
+        return serializer.serialize_error_to_response(error, operation, context.request.headers)
 
 
 class ServiceExceptionSerializer(ExceptionHandler):
@@ -225,7 +225,7 @@ def create_exception_response(self, exception: Exception, context: RequestContex
             context.service_exception = error
 
         serializer = create_serializer(context.service)  # TODO: serializer cache
-        return serializer.serialize_error_to_response(error, operation)
+        return serializer.serialize_error_to_response(error, operation, context.request.headers)
 
 
 class ServiceResponseParser(Handler):

diff --git a/localstack/aws/protocol/serializer.py b/localstack/aws/protocol/serializer.py
diff --git a/localstack/aws/skeleton.py b/localstack/aws/skeleton.py
@@ -171,7 +171,7 @@ def dispatch_request(self, context: RequestContext, instance: ServiceRequest) ->
         context.service_response = result
 
         # Serialize result dict to an HTTPResponse and return it
-        return self.serializer.serialize_to_response(result, operation)
+        return self.serializer.serialize_to_response(result, operation, context.request.headers)
 
     def on_service_exception(
         self, context: RequestContext, exception: ServiceException
@@ -185,7 +185,9 @@ def on_service_exception(
         """
         context.service_exception = exception
 
-        return self.serializer.serialize_error_to_response(exception, context.operation)
+        return self.serializer.serialize_error_to_response(
+            exception, context.operation, context.request.headers
+        )
 
     def on_not_implemented_error(self, context: RequestContext) -> HttpResponse:
         """
@@ -211,4 +213,4 @@ def on_not_implemented_error(self, context: RequestContext) -> HttpResponse:
         )
         context.service_exception = error
 
-        return serializer.serialize_error_to_response(error, operation)
+        return serializer.serialize_error_to_response(error, operation, context.request.headers)
diff --git a/localstack/config.py b/localstack/config.py
@@ -554,7 +554,7 @@ def in_docker():
 # Delay between data persistence (in seconds)
 KINESIS_MOCK_PERSIST_INTERVAL = os.environ.get("KINESIS_MOCK_PERSIST_INTERVAL", "").strip() or "5s"
 
-# Kinesis provider - either "kinesis-mock" or "kinesalite"
+# Kinesis provider - either "kinesis-mock" or "kinesalite" (deprecated, kinesalite support will be removed)
 KINESIS_PROVIDER = os.environ.get("KINESIS_PROVIDER") or "kinesis-mock"
 
 # Whether or not to handle lambda event sources as synchronous invocations

diff --git a/localstack/constants.py b/localstack/constants.py
@@ -90,6 +90,7 @@
 
 # content types / encodings
 HEADER_CONTENT_TYPE = "Content-Type"
+TEXT_XML = "text/xml"
 APPLICATION_AMZ_JSON_1_0 = "application/x-amz-json-1.0"
 APPLICATION_AMZ_JSON_1_1 = "application/x-amz-json-1.1"
 APPLICATION_AMZ_CBOR_1_1 = "application/x-amz-cbor-1.1"

diff --git a/localstack/services/kinesis/kinesis_starter.py b/localstack/services/kinesis/kinesis_starter.py
@@ -4,6 +4,7 @@
 from localstack import config
 from localstack.services.infra import log_startup_message, start_proxy_for_service
 from localstack.services.kinesis import kinesalite_server, kinesis_mock_server
+from localstack.services.plugins import SERVICE_PLUGINS
 from localstack.utils.aws import aws_stack
 from localstack.utils.serving import Server
 
@@ -38,12 +39,16 @@ def start_kinesis(
     _server.start()
     log_startup_message("Kinesis")
     port = port or config.service_port("kinesis")
-    start_proxy_for_service(
-        "kinesis",
-        port,
-        backend_port=_server.port,
-        update_listener=update_listener,
-    )
+    
+    # TODO: flip back to "!= kinesis:asf" to be sure we have the old control path when merging
+    if SERVICE_PLUGINS.get("kinesis").name() == "kinesis:legacy":
+        start_proxy_for_service(
+            "kinesis",
+            port,
+            backend_port=_server.port,
+            update_listener=update_listener,
+        )
+
     return _server
 
 
@@ -76,3 +81,7 @@ def is_kinesis_running() -> bool:
     if _server is None:
         return False
     return _server.is_running()
+
+
+def get_server():
+    return _server