<!-- 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 re
import os
import imp
import sys
import json
import uuid
import time
import base64
import hashlib
import logging
import functools
import threading
import traceback
from io import BytesIO
from datetime import datetime
from flask import Flask, Response, jsonify, request
from six.moves import cStringIO as StringIO
from six.moves.urllib.parse import urlparse
from moto.apigateway.models import apigateway_backends
from localstack import config
from localstack.constants import TEST_AWS_ACCOUNT_ID
from localstack.utils.aws import aws_stack, aws_responses
from localstack.utils.common import (