<!-- 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.
...

import os
import re
import json
import base64
import unittest
import xmltodict
from botocore.exceptions import ClientError
from jsonpatch import apply_patch
from requests.s import Response
from requests.structures import CaseInsensitiveDict
from localstack.utils import testutil
from localstack.utils.aws import aws_stack
from localstack.constants import TEST_AWS_ACCOUNT_ID
from localstack.utils.common import (
    to_str, json_safe, clone, short_uid, get_free_tcp_port, load_file, safe_requests as requests)
from localstack.services.generic_proxy import GenericProxy, ProxyListener
from localstack.services.apigateway.helpers import (
    get_rest_api_paths, get_resource_for_path, connect_api_gateway_to_sqs, gateway_request_url)
from localstack.services.awslambda.lambda_api import add_event_source
from localstack.services.awslambda.lambda_utils import LAMBDA_RUNTIME_PYTHON36
from .test_lambda import TEST_LAMBDA_PYTHON, TEST_LAMBDA_LIBS
THIS_FOLDER = os.path.dirname(os.path.realpath(__file__))
TEST_SWAGGER_FILE = os.path.join(THIS_FOLDER, 'files', 'swagger.json')