<!-- 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
I am trying to set up a localstack lambda function that has a locally-run kafka cluster as its event source. I am using aws cli v2. When I run the command to create event source mapping for my lambda function using aws cli, I get a "500 Internal Server Error".
...

## Expected behavior
A new lambda event source mapping should be created for my local kafka cluster.
...

## Actual behavior
Getting an Internal Server Error on aws cli output and getting a "KeyError: EventSourceArn" on localstack logs.
...

# Steps to reproduce
1. Created a lambda handler in python.
    ```python 3
    import json

    def handler(event, context):
        body = {
            "message": str(event)
        }
        response = {
            "statusCode": 200,
            "body": json.dumps(body)
        }
        return response

2. Created my lambda function.
    ```
    $  aws lambda create-function \
        --function-name fun1 \
        --handler lambda.handler \
        --runtime python3.8 \
        --role r1 \
        --zip-file fileb://l1.zip \
        --endpoint-url http://localhost:4566
    ```

3. Created a secret.
    ```
    aws secretsmanager create-secret --name localstack --endpoint-url http://localhost:4566
    ```

4. Created a kafka cluster and a topic called `t1` on my local system using the [official kafka binary](https://www.apache.org/dyn/closer.cgi?path=/kafka/2.7.0/kafka_2.13-2.7.0.tgz). It works without problems.

5. Tried to create event source mapping using the command mentioned [here](https://docs.aws.amazon.com/lambda/latest/dg/services-smaa-topic-add.html) in the official AWS website.
    ```
   $ aws lambda create-event-source-mapping \
      --topics t1 \
      --source-access-configuration Type=SASL_SCRAM_512_AUTH,URI=arn:aws:secretsmanager:us-east-1:000000000000:secret:localstack-TDIuI \
      --function-name arn:aws:lambda:us-east-1:000000000000:function:fun1 \
      --self-managed-event-source '{"Endpoints":{"KAFKA_BOOTSTRAP_SERVERS":["localhost:9092"]}}' \
      --endpoint-url http://localhost:4566
    ```
    
    Here, aws cli throws the following error message
    ```
    An error occurred (500) when calling the CreateEventSourceMapping operation (reached max retries: 2): <!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 3.2 Final//EN">
    <title>500 Internal Server Error</title>
    <h1>Internal Server Error</h1>
    <p>The server encountered an internal error and was unable to complete your request. Either the server is overloaded or there is an error in the application.</p>
    ```

    And the localstack log gives the following error message
    ```
    localstack_main | 2021-01-30T13:32:54:ERROR:lambda_api: Exception on /2015-03-31/event-source-mappings/ [POST]
    localstack_main | Traceback (most recent call last):
    localstack_main |   File "/opt/code/localstack/.venv/lib/python3.8/site-packages/flask/app.py", line 2447, in wsgi_app
    localstack_main |     response = self.full_dispatch_request()
    localstack_main |   File "/opt/code/localstack/.venv/lib/python3.8/site-packages/flask/app.py", line 1952, in full_dispatch_request
    localstack_main |     rv = self.handle_user_exception(e)
    localstack_main |   File "/opt/code/localstack/.venv/lib/python3.8/site-packages/flask_cors/extension.py", line 165, in wrapped_function
    localstack_main |     return cors_after_request(app.make_response(f(*args, **kwargs)))
    localstack_main |   File "/opt/code/localstack/.venv/lib/python3.8/site-packages/flask/app.py", line 1821, in handle_user_exception
    localstack_main |     reraise(exc_type, exc_value, tb)
    localstack_main |   File "/opt/code/localstack/.venv/lib/python3.8/site-packages/flask/_compat.py", line 39, in reraise
    localstack_main |     raise value
    localstack_main |   File "/opt/code/localstack/.venv/lib/python3.8/site-packages/flask/app.py", line 1950, in full_dispatch_request
    localstack_main |     rv = self.dispatch_request()
    localstack_main |   File "/opt/code/localstack/.venv/lib/python3.8/site-packages/flask/app.py", line 1936, in dispatch_request
    localstack_main |     return self.view_functions[rule.endpoint](**req.view_args)
    localstack_main |   File "/opt/code/localstack/localstack/services/awslambda/lambda_api.py", line 1457, in create_event_source_mapping
    localstack_main |     data['FunctionName'], data['EventSourceArn'], data.get('Enabled'), data.get('BatchSize')
    localstack_main | KeyError: 'EventSourceArn'
    ```

## Command used to start LocalStack
Copied the contents of docker-compose.yml from localstack repo as-is and ran the following command:
```
docker-compose up
```
...

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

self.assertEqual(response.status_code, 403)
        response = requests.put(
            url,
            json.dumps({'id': 'id1', 'data': 'foobar123'}),
            headers={'X-API-Key': api_key['value']}
        )
        # when 
        self.assertEqual(response.status_code, 200)
    def test_import_rest_api(self):
        rest_api_name = 'restapi-%s' % short_uid()
        client = aws_stack.connect_to_service('apigateway')
        rest_api_id = client.create_rest_api(name=rest_api_name)['id']
        spec_file = load_file(TEST_SWAGGER_FILE)
        rs = client.put_rest_api(
            restApiId=rest_api_id, body=spec_file)
        self.assertEqual(rs['ResponseMetadata']['HTTPStatusCode'], 200)
        rs = client.get_resources(restApiId=rest_api_id)
        self.assertEqual(len(rs['items']), 1)
        resource = rs['items'][0]
        self.assertEqual(resource['path'], '/test')
        self.assertIn('GET', resource['resourceMethods'])
        
        client.delete_rest_api(restApiId=rest_api_id)
    def test_step_function_integrations(self):
        client = aws_stack.connect_to_service('apigateway')
        sfn_client = aws_stack.connect_to_service('stepfunctions')
        lambda_client = aws_stack.connect_to_service('lambda')
        state_machine_name = 'test'
        state_machine_def = {
            'Comment': 'Hello World example',