Unable to decode kinesis event using a AWS Lambda written in Python

Question

I am pushing a simple payload of JSON using the spring-cloud-stream-binder-kinesis lib. The event works 100% and I am able to consume it from another micro service.

I am now trying to get a simple lambda up and running using Python 3.7. I have created a trigger off the stream and am receiving the event successfully (I can see the output in cloud watch logs as per below), however, when trying to decode the payload so I can access the JSON properties, it is appending some weird characters which is preventing me from being able to continue.

Here is the output of the logs after decoding the base64 item in the kinesis stream:

b'\xff\x01\x0bcontentType\x00\x00\x00\x12"application/json"
{
    "responseCode": 0,
    "responseMessage": null,
    "userId": 0,
    "emailAddress": "[email protected]",
    "name": "James",
    "dateRegistered": null,
    "socialNetwork": "Facebook"
}

Here is the simple Python handler:

def lambda_handler(event, context):
    
    for record in event['Records']:
       #Kinesis data is base64 encoded so decode here
       payload=base64.b64decode(record["kinesis"]["data"],validate=False)
       print("Payload" + str(payload))
    
   json_properties = json.loads(payload) <-- error here

Error:


[ERROR] UnicodeDecodeError: 'utf-8' codec can't decode byte 0xff in position 0: invalid start byte
Traceback (most recent call last):
  File "/var/task/Index.py", line 20, in lambda_handler
    json_properties = json.loads(payload)
  File "/var/lang/lib/python3.7/json/__init__.py", line 343, in loads
    s = s.decode(detect_encoding(s), 'surrogatepass')

Any help would be appreciated.

Philipp Johannis Philipp Johannis · Accepted Answer · 2020-10-29T19:33:55

Could you try the following, this is working in a python lambda I am using. I assume you are missing a utf-8 conversion:

from aws_kinesis_agg.deaggregator import deaggregate_records

def lambda_handler(event, context):
    
    raw_kinesis_records = event['Records']

    records = deaggregate_records(raw_kinesis_records)

    for record in records:
       #Kinesis data is base64 encoded so decode here
       payload=base64.b64decode(record["kinesis"]["data"],validate=False)
       payload_utf8=str(payload, 'utf-8')
       print("Payload" + payload_utf8)
    
   json_properties = json.loads(payload_utf8)

Unable to decode kinesis event using a AWS Lambda written in Python

1 Answers