We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Kinesis stream records do not offer AWS-defined optional values
<!-- Love localstack? Please consider supporting our collective:
👉  https://opencollective.com/localstack/donate -->

# Type of request: This is a ...

[ ] bug report
[x] feature request

# Detailed description

Kinesis does not populate/offer to populate all potential values in a record. I have a lambda being triggered by kinesis events. The records being passed in are only those required by AWS, but there are no optional values being used.

## Expected behavior

An event's kinesis object should be configurable to look as follows
```
{ 
    "ApproximateArrivalTimestamp": number,
    "Data": blob,
    "EncryptionType": "string",
    "PartitionKey": "string",
    "SequenceNumber": "string"
}
```

## Actual behavior

In its current state, only required values are present.
```
{ 
    "Data": blob,
    "PartitionKey": "string",
    "SequenceNumber": "string"
}
```

# Steps to reproduce

## Command used to start LocalStack

`docker-compose up` with given `docker-compose.yml`

## Client code (AWS SDK code snippet, or sequence of "awslocal" commands)

The source of this issue/feature request lies in the kinesis listener. AWS documentation of a kinesis stream record lists all values here: https://docs.aws.amazon.com/kinesis/latest/APIReference/API_Record.html

We are only returning the required values.
https://github.com/localstack/localstack/blob/78031dd65da9394f8b1b020be01ef02c63c433ee/localstack/services/kinesis/kinesis_listener.py#L85-L94

If there is no implicit reason for not offering optional values, I'm more than happy to submit a PR.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `localstack/services/kinesis/kinesis_listener.py`
Content:
```
1 import json
2 import random
3 from datetime import datetime
4 from requests.models import Response
5 from localstack import config
6 from localstack.utils.common import to_str, json_safe, clone
7 from localstack.utils.analytics import event_publisher
8 from localstack.services.awslambda import lambda_api
9 from localstack.services.generic_proxy import ProxyListener
10 
11 # action headers
12 ACTION_PREFIX = 'Kinesis_20131202'
13 ACTION_PUT_RECORD = '%s.PutRecord' % ACTION_PREFIX
14 ACTION_PUT_RECORDS = '%s.PutRecords' % ACTION_PREFIX
15 ACTION_LIST_STREAMS = '%s.ListStreams' % ACTION_PREFIX
16 ACTION_CREATE_STREAM = '%s.CreateStream' % ACTION_PREFIX
17 ACTION_DELETE_STREAM = '%s.DeleteStream' % ACTION_PREFIX
18 ACTION_UPDATE_SHARD_COUNT = '%s.UpdateShardCount' % ACTION_PREFIX
19 
20 # list of stream consumer details
21 STREAM_CONSUMERS = []
22 
23 
24 class ProxyListenerKinesis(ProxyListener):
25 
26     def forward_request(self, method, path, data, headers):
27         global STREAM_CONSUMERS
28         data = json.loads(to_str(data or '{}'))
29         action = headers.get('X-Amz-Target')
30 
31         if action == '%s.RegisterStreamConsumer' % ACTION_PREFIX:
32             consumer = clone(data)
33             consumer['ConsumerStatus'] = 'ACTIVE'
34             consumer['ConsumerARN'] = '%s/consumer/%s' % (data['StreamARN'], data['ConsumerName'])
35             consumer['ConsumerCreationTimestamp'] = datetime.now()
36             consumer = json_safe(consumer)
37             STREAM_CONSUMERS.append(consumer)
38             return {'Consumer': consumer}
39         elif action == '%s.DeregisterStreamConsumer' % ACTION_PREFIX:
40             def consumer_matches(c):
41                 stream_arn = data.get('StreamARN')
42                 cons_name = data.get('ConsumerName')
43                 cons_arn = data.get('ConsumerARN')
44                 return (c.get('ConsumerARN') == cons_arn or
45                     (c.get('StreamARN') == stream_arn and c.get('ConsumerName') == cons_name))
46             STREAM_CONSUMERS = [c for c in STREAM_CONSUMERS if not consumer_matches(c)]
47             return {}
48         elif action == '%s.ListStreamConsumers' % ACTION_PREFIX:
49             result = {
50                 'Consumers': [c for c in STREAM_CONSUMERS if c.get('StreamARN') == data.get('StreamARN')]
51             }
52             return result
53         elif action == '%s.DescribeStreamConsumer' % ACTION_PREFIX:
54             consumer_arn = data.get('ConsumerARN') or data['ConsumerName']
55             consumer_name = data.get('ConsumerName') or data['ConsumerARN']
56             result = {
57                 'ConsumerDescription': {
58                     'ConsumerARN': consumer_arn,
59                     # 'ConsumerCreationTimestamp': number,
60                     'ConsumerName': consumer_name,
61                     'ConsumerStatus': 'ACTIVE',
62                     'StreamARN': data.get('StreamARN')
63                 }
64             }
65             return result
66 
67         if random.random() < config.KINESIS_ERROR_PROBABILITY:
68             action = headers.get('X-Amz-Target')
69             if action in [ACTION_PUT_RECORD, ACTION_PUT_RECORDS]:
70                 return kinesis_error_response(data, action)
71         return True
72 
73     def return_response(self, method, path, data, headers, response):
74         action = headers.get('X-Amz-Target')
75         data = json.loads(to_str(data or '{}'))
76 
77         records = []
78         if action in (ACTION_CREATE_STREAM, ACTION_DELETE_STREAM):
79             event_type = (event_publisher.EVENT_KINESIS_CREATE_STREAM if action == ACTION_CREATE_STREAM
80                           else event_publisher.EVENT_KINESIS_DELETE_STREAM)
81             payload = {'n': event_publisher.get_hash(data.get('StreamName'))}
82             if action == ACTION_CREATE_STREAM:
83                 payload['s'] = data.get('ShardCount')
84             event_publisher.fire_event(event_type, payload=payload)
85         elif action == ACTION_PUT_RECORD:
86             response_body = json.loads(to_str(response.content))
87             event_record = {
88                 'data': data['Data'],
89                 'partitionKey': data['PartitionKey'],
90                 'sequenceNumber': response_body.get('SequenceNumber')
91             }
92             event_records = [event_record]
93             stream_name = data['StreamName']
94             lambda_api.process_kinesis_records(event_records, stream_name)
95         elif action == ACTION_PUT_RECORDS:
96             event_records = []
97             response_body = json.loads(to_str(response.content))
98             if 'Records' in response_body:
99                 response_records = response_body['Records']
100                 records = data['Records']
101                 for index in range(0, len(records)):
102                     record = records[index]
103                     event_record = {
104                         'data': record['Data'],
105                         'partitionKey': record['PartitionKey'],
106                         'sequenceNumber': response_records[index].get('SequenceNumber')
107                     }
108                     event_records.append(event_record)
109                 stream_name = data['StreamName']
110                 lambda_api.process_kinesis_records(event_records, stream_name)
111         elif action == ACTION_UPDATE_SHARD_COUNT:
112             # Currently kinesalite, which backs the Kinesis implementation for localstack, does
113             # not support UpdateShardCount:
114             # https://github.com/mhart/kinesalite/issues/61
115             #
116             # [Terraform](https://www.terraform.io) makes the call to UpdateShardCount when it
117             # applies Kinesis resources. A Terraform run fails when this is not present.
118             #
119             # The code that follows just returns a successful response, bypassing the 400
120             # response that kinesalite returns.
121             #
122             response = Response()
123             response.status_code = 200
124             content = {
125                 'CurrentShardCount': 1,
126                 'StreamName': data['StreamName'],
127                 'TargetShardCount': data['TargetShardCount']
128             }
129             response.encoding = 'UTF-8'
130             response._content = json.dumps(content)
131             return response
132 
133 
134 # instantiate listener
135 UPDATE_KINESIS = ProxyListenerKinesis()
136 
137 
138 def kinesis_error_response(data, action):
139     error_response = Response()
140 
141     if action == ACTION_PUT_RECORD:
142         error_response.status_code = 400
143         content = {
144             'ErrorCode': 'ProvisionedThroughputExceededException',
145             'ErrorMessage': 'Rate exceeded for shard X in stream Y under account Z.'
146         }
147     else:
148         error_response.status_code = 200
149         content = {'FailedRecordCount': 1, 'Records': []}
150         for record in data.get('Records', []):
151             content['Records'].append({
152                 'ErrorCode': 'ProvisionedThroughputExceededException',
153                 'ErrorMessage': 'Rate exceeded for shard X in stream Y under account Z.'
154             })
155 
156     error_response._content = json.dumps(content)
157     return error_response
158
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/localstack/services/kinesis/kinesis_listener.py b/localstack/services/kinesis/kinesis_listener.py
--- a/localstack/services/kinesis/kinesis_listener.py
+++ b/localstack/services/kinesis/kinesis_listener.py
@@ -1,9 +1,8 @@
 import json
 import random
-from datetime import datetime
 from requests.models import Response
 from localstack import config
-from localstack.utils.common import to_str, json_safe, clone
+from localstack.utils.common import to_str, json_safe, clone, timestamp_millis
 from localstack.utils.analytics import event_publisher
 from localstack.services.awslambda import lambda_api
 from localstack.services.generic_proxy import ProxyListener
@@ -32,7 +31,7 @@
             consumer = clone(data)
             consumer['ConsumerStatus'] = 'ACTIVE'
             consumer['ConsumerARN'] = '%s/consumer/%s' % (data['StreamARN'], data['ConsumerName'])
-            consumer['ConsumerCreationTimestamp'] = datetime.now()
+            consumer['ConsumerCreationTimestamp'] = timestamp_millis()
             consumer = json_safe(consumer)
             STREAM_CONSUMERS.append(consumer)
             return {'Consumer': consumer}
@@ -85,7 +84,9 @@
         elif action == ACTION_PUT_RECORD:
             response_body = json.loads(to_str(response.content))
             event_record = {
+                'approximateArrivalTimestamp': timestamp_millis(),
                 'data': data['Data'],
+                'encryptionType': 'NONE',
                 'partitionKey': data['PartitionKey'],
                 'sequenceNumber': response_body.get('SequenceNumber')
             }
@@ -101,7 +102,9 @@
                 for index in range(0, len(records)):
                     record = records[index]
                     event_record = {
+                        'approximateArrivalTimestamp': timestamp_millis(),
                         'data': record['Data'],
+                        'encryptionType': 'NONE',
                         'partitionKey': record['PartitionKey'],
                         'sequenceNumber': response_records[index].get('SequenceNumber')
                     }

{"golden_diff": "diff --git a/localstack/services/kinesis/kinesis_listener.py b/localstack/services/kinesis/kinesis_listener.py\n--- a/localstack/services/kinesis/kinesis_listener.py\n+++ b/localstack/services/kinesis/kinesis_listener.py\n@@ -1,9 +1,8 @@\n import json\n import random\n-from datetime import datetime\n from requests.models import Response\n from localstack import config\n-from localstack.utils.common import to_str, json_safe, clone\n+from localstack.utils.common import to_str, json_safe, clone, timestamp_millis\n from localstack.utils.analytics import event_publisher\n from localstack.services.awslambda import lambda_api\n from localstack.services.generic_proxy import ProxyListener\n@@ -32,7 +31,7 @@\n             consumer = clone(data)\n             consumer['ConsumerStatus'] = 'ACTIVE'\n             consumer['ConsumerARN'] = '%s/consumer/%s' % (data['StreamARN'], data['ConsumerName'])\n-            consumer['ConsumerCreationTimestamp'] = datetime.now()\n+            consumer['ConsumerCreationTimestamp'] = timestamp_millis()\n             consumer = json_safe(consumer)\n             STREAM_CONSUMERS.append(consumer)\n             return {'Consumer': consumer}\n@@ -85,7 +84,9 @@\n         elif action == ACTION_PUT_RECORD:\n             response_body = json.loads(to_str(response.content))\n             event_record = {\n+                'approximateArrivalTimestamp': timestamp_millis(),\n                 'data': data['Data'],\n+                'encryptionType': 'NONE',\n                 'partitionKey': data['PartitionKey'],\n                 'sequenceNumber': response_body.get('SequenceNumber')\n             }\n@@ -101,7 +102,9 @@\n                 for index in range(0, len(records)):\n                     record = records[index]\n                     event_record = {\n+                        'approximateArrivalTimestamp': timestamp_millis(),\n                         'data': record['Data'],\n+                        'encryptionType': 'NONE',\n                         'partitionKey': record['PartitionKey'],\n                         'sequenceNumber': response_records[index].get('SequenceNumber')\n                     }\n", "issue": "Kinesis stream records do not offer AWS-defined optional values\n<!-- Love localstack? Please consider supporting our collective:\r\n\ud83d\udc49  https://opencollective.com/localstack/donate -->\r\n\r\n# Type of request: This is a ...\r\n\r\n[ ] bug report\r\n[x] feature request\r\n\r\n# Detailed description\r\n\r\nKinesis does not populate/offer to populate all potential values in a record. I have a lambda being triggered by kinesis events. The records being passed in are only those required by AWS, but there are no optional values being used.\r\n\r\n## Expected behavior\r\n\r\nAn event's kinesis object should be configurable to look as follows\r\n```\r\n{ \r\n    \"ApproximateArrivalTimestamp\": number,\r\n    \"Data\": blob,\r\n    \"EncryptionType\": \"string\",\r\n    \"PartitionKey\": \"string\",\r\n    \"SequenceNumber\": \"string\"\r\n}\r\n```\r\n\r\n## Actual behavior\r\n\r\nIn its current state, only required values are present.\r\n```\r\n{ \r\n    \"Data\": blob,\r\n    \"PartitionKey\": \"string\",\r\n    \"SequenceNumber\": \"string\"\r\n}\r\n```\r\n\r\n# Steps to reproduce\r\n\r\n## Command used to start LocalStack\r\n\r\n`docker-compose up` with given `docker-compose.yml`\r\n\r\n## Client code (AWS SDK code snippet, or sequence of \"awslocal\" commands)\r\n\r\nThe source of this issue/feature request lies in the kinesis listener. AWS documentation of a kinesis stream record lists all values here: https://docs.aws.amazon.com/kinesis/latest/APIReference/API_Record.html\r\n\r\nWe are only returning the required values.\r\nhttps://github.com/localstack/localstack/blob/78031dd65da9394f8b1b020be01ef02c63c433ee/localstack/services/kinesis/kinesis_listener.py#L85-L94\r\n\r\nIf there is no implicit reason for not offering optional values, I'm more than happy to submit a PR.\r\n\n", "before_files": [{"content": "import json\nimport random\nfrom datetime import datetime\nfrom requests.models import Response\nfrom localstack import config\nfrom localstack.utils.common import to_str, json_safe, clone\nfrom localstack.utils.analytics import event_publisher\nfrom localstack.services.awslambda import lambda_api\nfrom localstack.services.generic_proxy import ProxyListener\n\n# action headers\nACTION_PREFIX = 'Kinesis_20131202'\nACTION_PUT_RECORD = '%s.PutRecord' % ACTION_PREFIX\nACTION_PUT_RECORDS = '%s.PutRecords' % ACTION_PREFIX\nACTION_LIST_STREAMS = '%s.ListStreams' % ACTION_PREFIX\nACTION_CREATE_STREAM = '%s.CreateStream' % ACTION_PREFIX\nACTION_DELETE_STREAM = '%s.DeleteStream' % ACTION_PREFIX\nACTION_UPDATE_SHARD_COUNT = '%s.UpdateShardCount' % ACTION_PREFIX\n\n# list of stream consumer details\nSTREAM_CONSUMERS = []\n\n\nclass ProxyListenerKinesis(ProxyListener):\n\n    def forward_request(self, method, path, data, headers):\n        global STREAM_CONSUMERS\n        data = json.loads(to_str(data or '{}'))\n        action = headers.get('X-Amz-Target')\n\n        if action == '%s.RegisterStreamConsumer' % ACTION_PREFIX:\n            consumer = clone(data)\n            consumer['ConsumerStatus'] = 'ACTIVE'\n            consumer['ConsumerARN'] = '%s/consumer/%s' % (data['StreamARN'], data['ConsumerName'])\n            consumer['ConsumerCreationTimestamp'] = datetime.now()\n            consumer = json_safe(consumer)\n            STREAM_CONSUMERS.append(consumer)\n            return {'Consumer': consumer}\n        elif action == '%s.DeregisterStreamConsumer' % ACTION_PREFIX:\n            def consumer_matches(c):\n                stream_arn = data.get('StreamARN')\n                cons_name = data.get('ConsumerName')\n                cons_arn = data.get('ConsumerARN')\n                return (c.get('ConsumerARN') == cons_arn or\n                    (c.get('StreamARN') == stream_arn and c.get('ConsumerName') == cons_name))\n            STREAM_CONSUMERS = [c for c in STREAM_CONSUMERS if not consumer_matches(c)]\n            return {}\n        elif action == '%s.ListStreamConsumers' % ACTION_PREFIX:\n            result = {\n                'Consumers': [c for c in STREAM_CONSUMERS if c.get('StreamARN') == data.get('StreamARN')]\n            }\n            return result\n        elif action == '%s.DescribeStreamConsumer' % ACTION_PREFIX:\n            consumer_arn = data.get('ConsumerARN') or data['ConsumerName']\n            consumer_name = data.get('ConsumerName') or data['ConsumerARN']\n            result = {\n                'ConsumerDescription': {\n                    'ConsumerARN': consumer_arn,\n                    # 'ConsumerCreationTimestamp': number,\n                    'ConsumerName': consumer_name,\n                    'ConsumerStatus': 'ACTIVE',\n                    'StreamARN': data.get('StreamARN')\n                }\n            }\n            return result\n\n        if random.random() < config.KINESIS_ERROR_PROBABILITY:\n            action = headers.get('X-Amz-Target')\n            if action in [ACTION_PUT_RECORD, ACTION_PUT_RECORDS]:\n                return kinesis_error_response(data, action)\n        return True\n\n    def return_response(self, method, path, data, headers, response):\n        action = headers.get('X-Amz-Target')\n        data = json.loads(to_str(data or '{}'))\n\n        records = []\n        if action in (ACTION_CREATE_STREAM, ACTION_DELETE_STREAM):\n            event_type = (event_publisher.EVENT_KINESIS_CREATE_STREAM if action == ACTION_CREATE_STREAM\n                          else event_publisher.EVENT_KINESIS_DELETE_STREAM)\n            payload = {'n': event_publisher.get_hash(data.get('StreamName'))}\n            if action == ACTION_CREATE_STREAM:\n                payload['s'] = data.get('ShardCount')\n            event_publisher.fire_event(event_type, payload=payload)\n        elif action == ACTION_PUT_RECORD:\n            response_body = json.loads(to_str(response.content))\n            event_record = {\n                'data': data['Data'],\n                'partitionKey': data['PartitionKey'],\n                'sequenceNumber': response_body.get('SequenceNumber')\n            }\n            event_records = [event_record]\n            stream_name = data['StreamName']\n            lambda_api.process_kinesis_records(event_records, stream_name)\n        elif action == ACTION_PUT_RECORDS:\n            event_records = []\n            response_body = json.loads(to_str(response.content))\n            if 'Records' in response_body:\n                response_records = response_body['Records']\n                records = data['Records']\n                for index in range(0, len(records)):\n                    record = records[index]\n                    event_record = {\n                        'data': record['Data'],\n                        'partitionKey': record['PartitionKey'],\n                        'sequenceNumber': response_records[index].get('SequenceNumber')\n                    }\n                    event_records.append(event_record)\n                stream_name = data['StreamName']\n                lambda_api.process_kinesis_records(event_records, stream_name)\n        elif action == ACTION_UPDATE_SHARD_COUNT:\n            # Currently kinesalite, which backs the Kinesis implementation for localstack, does\n            # not support UpdateShardCount:\n            # https://github.com/mhart/kinesalite/issues/61\n            #\n            # [Terraform](https://www.terraform.io) makes the call to UpdateShardCount when it\n            # applies Kinesis resources. A Terraform run fails when this is not present.\n            #\n            # The code that follows just returns a successful response, bypassing the 400\n            # response that kinesalite returns.\n            #\n            response = Response()\n            response.status_code = 200\n            content = {\n                'CurrentShardCount': 1,\n                'StreamName': data['StreamName'],\n                'TargetShardCount': data['TargetShardCount']\n            }\n            response.encoding = 'UTF-8'\n            response._content = json.dumps(content)\n            return response\n\n\n# instantiate listener\nUPDATE_KINESIS = ProxyListenerKinesis()\n\n\ndef kinesis_error_response(data, action):\n    error_response = Response()\n\n    if action == ACTION_PUT_RECORD:\n        error_response.status_code = 400\n        content = {\n            'ErrorCode': 'ProvisionedThroughputExceededException',\n            'ErrorMessage': 'Rate exceeded for shard X in stream Y under account Z.'\n        }\n    else:\n        error_response.status_code = 200\n        content = {'FailedRecordCount': 1, 'Records': []}\n        for record in data.get('Records', []):\n            content['Records'].append({\n                'ErrorCode': 'ProvisionedThroughputExceededException',\n                'ErrorMessage': 'Rate exceeded for shard X in stream Y under account Z.'\n            })\n\n    error_response._content = json.dumps(content)\n    return error_response\n", "path": "localstack/services/kinesis/kinesis_listener.py"}], "after_files": [{"content": "import json\nimport random\nfrom requests.models import Response\nfrom localstack import config\nfrom localstack.utils.common import to_str, json_safe, clone, timestamp_millis\nfrom localstack.utils.analytics import event_publisher\nfrom localstack.services.awslambda import lambda_api\nfrom localstack.services.generic_proxy import ProxyListener\n\n# action headers\nACTION_PREFIX = 'Kinesis_20131202'\nACTION_PUT_RECORD = '%s.PutRecord' % ACTION_PREFIX\nACTION_PUT_RECORDS = '%s.PutRecords' % ACTION_PREFIX\nACTION_LIST_STREAMS = '%s.ListStreams' % ACTION_PREFIX\nACTION_CREATE_STREAM = '%s.CreateStream' % ACTION_PREFIX\nACTION_DELETE_STREAM = '%s.DeleteStream' % ACTION_PREFIX\nACTION_UPDATE_SHARD_COUNT = '%s.UpdateShardCount' % ACTION_PREFIX\n\n# list of stream consumer details\nSTREAM_CONSUMERS = []\n\n\nclass ProxyListenerKinesis(ProxyListener):\n\n    def forward_request(self, method, path, data, headers):\n        global STREAM_CONSUMERS\n        data = json.loads(to_str(data or '{}'))\n        action = headers.get('X-Amz-Target')\n\n        if action == '%s.RegisterStreamConsumer' % ACTION_PREFIX:\n            consumer = clone(data)\n            consumer['ConsumerStatus'] = 'ACTIVE'\n            consumer['ConsumerARN'] = '%s/consumer/%s' % (data['StreamARN'], data['ConsumerName'])\n            consumer['ConsumerCreationTimestamp'] = timestamp_millis()\n            consumer = json_safe(consumer)\n            STREAM_CONSUMERS.append(consumer)\n            return {'Consumer': consumer}\n        elif action == '%s.DeregisterStreamConsumer' % ACTION_PREFIX:\n            def consumer_matches(c):\n                stream_arn = data.get('StreamARN')\n                cons_name = data.get('ConsumerName')\n                cons_arn = data.get('ConsumerARN')\n                return (c.get('ConsumerARN') == cons_arn or\n                    (c.get('StreamARN') == stream_arn and c.get('ConsumerName') == cons_name))\n            STREAM_CONSUMERS = [c for c in STREAM_CONSUMERS if not consumer_matches(c)]\n            return {}\n        elif action == '%s.ListStreamConsumers' % ACTION_PREFIX:\n            result = {\n                'Consumers': [c for c in STREAM_CONSUMERS if c.get('StreamARN') == data.get('StreamARN')]\n            }\n            return result\n        elif action == '%s.DescribeStreamConsumer' % ACTION_PREFIX:\n            consumer_arn = data.get('ConsumerARN') or data['ConsumerName']\n            consumer_name = data.get('ConsumerName') or data['ConsumerARN']\n            result = {\n                'ConsumerDescription': {\n                    'ConsumerARN': consumer_arn,\n                    # 'ConsumerCreationTimestamp': number,\n                    'ConsumerName': consumer_name,\n                    'ConsumerStatus': 'ACTIVE',\n                    'StreamARN': data.get('StreamARN')\n                }\n            }\n            return result\n\n        if random.random() < config.KINESIS_ERROR_PROBABILITY:\n            action = headers.get('X-Amz-Target')\n            if action in [ACTION_PUT_RECORD, ACTION_PUT_RECORDS]:\n                return kinesis_error_response(data, action)\n        return True\n\n    def return_response(self, method, path, data, headers, response):\n        action = headers.get('X-Amz-Target')\n        data = json.loads(to_str(data or '{}'))\n\n        records = []\n        if action in (ACTION_CREATE_STREAM, ACTION_DELETE_STREAM):\n            event_type = (event_publisher.EVENT_KINESIS_CREATE_STREAM if action == ACTION_CREATE_STREAM\n                          else event_publisher.EVENT_KINESIS_DELETE_STREAM)\n            payload = {'n': event_publisher.get_hash(data.get('StreamName'))}\n            if action == ACTION_CREATE_STREAM:\n                payload['s'] = data.get('ShardCount')\n            event_publisher.fire_event(event_type, payload=payload)\n        elif action == ACTION_PUT_RECORD:\n            response_body = json.loads(to_str(response.content))\n            event_record = {\n                'approximateArrivalTimestamp': timestamp_millis(),\n                'data': data['Data'],\n                'encryptionType': 'NONE',\n                'partitionKey': data['PartitionKey'],\n                'sequenceNumber': response_body.get('SequenceNumber')\n            }\n            event_records = [event_record]\n            stream_name = data['StreamName']\n            lambda_api.process_kinesis_records(event_records, stream_name)\n        elif action == ACTION_PUT_RECORDS:\n            event_records = []\n            response_body = json.loads(to_str(response.content))\n            if 'Records' in response_body:\n                response_records = response_body['Records']\n                records = data['Records']\n                for index in range(0, len(records)):\n                    record = records[index]\n                    event_record = {\n                        'approximateArrivalTimestamp': timestamp_millis(),\n                        'data': record['Data'],\n                        'encryptionType': 'NONE',\n                        'partitionKey': record['PartitionKey'],\n                        'sequenceNumber': response_records[index].get('SequenceNumber')\n                    }\n                    event_records.append(event_record)\n                stream_name = data['StreamName']\n                lambda_api.process_kinesis_records(event_records, stream_name)\n        elif action == ACTION_UPDATE_SHARD_COUNT:\n            # Currently kinesalite, which backs the Kinesis implementation for localstack, does\n            # not support UpdateShardCount:\n            # https://github.com/mhart/kinesalite/issues/61\n            #\n            # [Terraform](https://www.terraform.io) makes the call to UpdateShardCount when it\n            # applies Kinesis resources. A Terraform run fails when this is not present.\n            #\n            # The code that follows just returns a successful response, bypassing the 400\n            # response that kinesalite returns.\n            #\n            response = Response()\n            response.status_code = 200\n            content = {\n                'CurrentShardCount': 1,\n                'StreamName': data['StreamName'],\n                'TargetShardCount': data['TargetShardCount']\n            }\n            response.encoding = 'UTF-8'\n            response._content = json.dumps(content)\n            return response\n\n\n# instantiate listener\nUPDATE_KINESIS = ProxyListenerKinesis()\n\n\ndef kinesis_error_response(data, action):\n    error_response = Response()\n\n    if action == ACTION_PUT_RECORD:\n        error_response.status_code = 400\n        content = {\n            'ErrorCode': 'ProvisionedThroughputExceededException',\n            'ErrorMessage': 'Rate exceeded for shard X in stream Y under account Z.'\n        }\n    else:\n        error_response.status_code = 200\n        content = {'FailedRecordCount': 1, 'Records': []}\n        for record in data.get('Records', []):\n            content['Records'].append({\n                'ErrorCode': 'ProvisionedThroughputExceededException',\n                'ErrorMessage': 'Rate exceeded for shard X in stream Y under account Z.'\n            })\n\n    error_response._content = json.dumps(content)\n    return error_response\n", "path": "localstack/services/kinesis/kinesis_listener.py"}]}