# Type of request: This is a ...

[x] bug report
[ ] feature request

# Detailed description

We have two producers, one node and one JVM. Producing from the node one is using the official AWS SDK Kinesis client and simply puts the record as serialized JSON:

`client.putRecord({ Data: JSON.stringify(data), StreamName: topic, PartitionKey: '1' })`

This works fine and we're able to consume these records in other services fine with localstack.

The JVM application produces records using the [amazon-kinesis-client](https://github.com/awslabs/amazon-kinesis-client) library. whose `putRecord` requires the `data` has to be of type `software.amazon.awssdk.core.SdkBytes`, from the docs: 

> The data blob to put into the record, which is base64-encoded when the blob is serialized. When the data blob (the payload before base64-encoding) is added to the partition key size, the total size must not exceed the maximum record size (1 MiB).

Our code (Kotlin):

```
kinesisClient.putRecord {
    it.streamName(...)
      .partitionKey(...)
      .data(SdkBytes.fromByteArray(objectMapper.writeValueAsBytes(data)))
}
```

Both of these two producers are able to successfully produce records, and towards AWS proper it works as expected. However when trying to read the records back from localstack, the client is met with a `500 Unknown` error. I've verified that this happens both with the AWS CLI and a Node-client library.

The stack trace in localstack following this error:
```
2021-02-10T10:01:10:WARNING:localstack.utils.server.http2_server: Error in proxy handler for request POST http://localhost:4566/: Object of type bytearray is not JSON serializable Traceback (most recent call last):
  File "/opt/code/localstack/localstack/utils/server/http2_server.py", line 107, in index
    raise result
  File "/opt/code/localstack/localstack/utils/bootstrap.py", line 583, in run
    result = self.func(self.params)
  File "/opt/code/localstack/localstack/utils/async_utils.py", line 28, in _run
    return fn(*args, **kwargs)
  File "/opt/code/localstack/localstack/services/generic_proxy.py", line 605, in handler
    response = modify_and_forward(method=method, path=path_with_params, data_bytes=data, headers=headers,
  File "/opt/code/localstack/localstack/services/generic_proxy.py", line 373, in modify_and_forward
    listener_result = listener.forward_request(method=method,
  File "/opt/code/localstack/localstack/services/edge.py", line 105, in forward_request
    return do_forward_request(api, method, path, data, headers, port=port)
  File "/opt/code/localstack/localstack/services/edge.py", line 116, in do_forward_request
    result = do_forward_request_inmem(api, method, path, data, headers, port=port)
  File "/opt/code/localstack/localstack/services/edge.py", line 136, in do_forward_request_inmem
    response = modify_and_forward(method=method, path=path, data_bytes=data, headers=headers,
  File "/opt/code/localstack/localstack/services/generic_proxy.py", line 441, in modify_and_forward
    updated_response = update_listener.return_response(**kwargs)
  File "/opt/code/localstack/localstack/services/kinesis/kinesis_listener.py", line 158, in return_response
    response._content = json.dumps(results)
  File "/usr/lib/python3.8/json/__init__.py", line 231, in dumps
    return _default_encoder.encode(obj)
  File "/usr/lib/python3.8/json/encoder.py", line 199, in encode
    chunks = self.iterencode(o, _one_shot=True)
  File "/usr/lib/python3.8/json/encoder.py", line 257, in iterencode
    return _iterencode(o, 0)
  File "/usr/lib/python3.8/json/encoder.py", line 179, in default
    raise TypeError(f'Object of type {o.__class__.__name__} '
TypeError: Object of type bytearray is not JSON serializable
```


## Expected behavior

I'm able to read out the record as normal.

## Actual behavior

localstack crashes when record is read.

# Steps to reproduce

1. Write a record to a stream using https://github.com/awslabs/amazon-kinesis-client on the JVM
2. Read record using AWS CLI/Any other library

## Command used to start LocalStack

Using docker-compose:

```
version: '2.1'

services:
  docker-host:
    image: qoomon/docker-host
    cap_add: [ 'NET_ADMIN', 'NET_RAW' ]
    mem_limit: 8M
    restart: on-failure
  localstack:
    container_name: "${LOCALSTACK_DOCKER_NAME-localstack-dgv_main}"
    image: localstack/localstack:0.12.6
    depends_on: [ docker-host ]
    ports:
      - "4566:4566"
      - "4571:4571"
    environment:
      - DOCKER_HOST=unix:///var/run/docker.sock
      - HOST_TMP_FOLDER=${TMPDIR}
      - DATA_DIR=${TMPDIR}/data
      - DEFAULT_REGION=eu-west-1
      - LAMBDA_REMOTE_DOCKER=0
      - LAMBDA_EXECUTOR=docker-reuse
      - LAMBDA_DOCKER_NETWORK=scripts_default
      - AWS_CBOR_DISABLE=true
    volumes:
      - "${TMPDIR:-/tmp/localstack}:/tmp/localstack"
      - "/var/run/docker.sock:/var/run/docker.sock"
```

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

awslocal command for pulling records out from stream:

```
awslocal kinesis get-records --shard-iterator $(awslocal kinesis get-shard-iterator --stream-name test-stream-name --shard-id $(awslocal kinesis describe-stream --stream-name test-stream-name --query 'StreamDescription.Shards[0].ShardId' --output text) --shard-iterator-type AT_SEQUENCE_NUMBER --starting-sequence-number $(awslocal kinesis describe-stream --stream-name test-stream-name --query 'StreamDescription.Shards[0].SequenceNumberRange.StartingSequenceNumber' --output text) --query ShardIterator --output text)
```

if re.search(r'\/2.\d+.\d+', user_agent):
            return True
        return False
    def replace_in_encoded(self, data):
        if not data:
            return ''
        def _replace(_data):
            return re.sub(r'arn:aws:kinesis:[^:]+:', 'arn:aws:kinesis:%s:' % aws_stack.get_region(), _data)
        decoded, type_encoding = self.decode_content(data, True)
        if type_encoding == APPLICATION_JSON:
            return _replace(to_str(data))
        if type_encoding == APPLICATION_CBOR:
            replaced = _replace(json.dumps(decoded))
            return cbor2.dumps(json.loads(replaced))
    def decode_content(self, data, describe=False):
        content_type = ''
        try:
            decoded = json.loads(to_str(data))
            content_type = APPLICATION_JSON
        except UnicodeDecodeError:
            decoded = cbor2.loads(data)
            content_type = APPLICATION_CBOR
        if describe:
            return decoded, content_type
        return decoded
def subscribe_to_shard(data):